推荐系统

rebalance/base_signal.py

@@ -52,7 +52,7 @@ class BaseRebalanceSignal(RebalanceSignal, ABC):
             date = pd.to_datetime(signal['date'].replace(day=transfer_date))
             # 说明发生了跨月份问题
             if signal['date'].day > transfer_date:
-                if rrs.get_count(risk=PortfoliosRisk.FT3, effective=True) > 1:
+                if rrs.get_count(risk=PortfoliosRisk.FT3, effective=True) > 0:
                     date = date + pd.DateOffset(months=1)
             date = date + pd.DateOffset(months=frequency)
             date = date - timedelta(days=1)

rebalance/dao/robo_rebalance_signal.py

@@ -10,6 +10,7 @@ __COLUMNS__ = {
     'rrs_p_type': 'portfolio_type',
     'rrs_p_weight': 'portfolio',
     'rrs_effective': 'effective',
+    'rrs_create_time': 'create_time',
 }
 
 

web/robo_controller.py

 import datetime as dt
 import json
-from multiprocessing import Process
+import logging
+from statistics import pstdev
 
+import pandas as pd
 import uvicorn
-from apscheduler.schedulers.blocking import BlockingScheduler
-from fastapi import FastAPI
-from py_jftech import prev_workday, filter_weekend, autowired
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.date import DateTrigger
+from empyrical import sharpe_ratio, annual_volatility
+from fastapi import FastAPI, Request
+from py_jftech import prev_workday, filter_weekend, autowired, next_workday
+from starlette.responses import JSONResponse
 
 import main
-from api import DatumType, PortfoliosRisk, Datum
+from api import DatumType, PortfoliosRisk, Datum, RoboReportor
 
 app = FastAPI()
+# 创建 AsyncIOScheduler 实例
+scheduler = AsyncIOScheduler()
 
 REC_GID = 'E3886FBA-123B-7890-123E-123456BEEED'
+fund_infos, cp, roi, risk = None, None, None, None
 
 
 def get_today_rec():
@@ -33,32 +41,41 @@ def get_last_signal():
 
 @autowired
 def get_fund_infos(datum: Datum = None):
-    return datum.get_datums(DatumType.FUND)
+    global fund_infos
+    fund_infos = datum.get_datums(DatumType.FUND)
 
 
-@app.get("/recommend")
-async def root():
-    portfolio = get_today_rec()
-    if portfolio:
-        fund_infos = get_fund_infos()
+@autowired(names={'combo': 'hold-report'})
+def load_report(max_date=prev_workday(dt.date.today()), min_date=None, combo: RoboReportor = None):
+    global cp, roi, risk
+    datas = pd.DataFrame(combo.load_report(max_date=max_date, min_date=min_date))
+    datas.set_index('date', inplace=True)
+    datas = datas['acc_av']
+    returns = round(datas.pct_change(), 5)
+    cp = round(sharpe_ratio(returns, risk_free=0, period='daily', annualization=None), 2)
+    roi = round(annual_volatility(datas), 1)
+    risk = round(pstdev(datas), 1)
+    return cp, roi, risk
+
+
+@app.get("/franklin/prediction_data")
+async def recommend():
+    sig = get_last_signal()
+    if sig:
+        if not fund_infos:
+            get_fund_infos()
         id_ticker_map = {str(info['id']): info for info in fund_infos}
-        funds = json.loads(portfolio['portfolio'])
+        funds = json.loads(sig['portfolio'])
         rec_list = []
         portfolios = {'recomm_guid': REC_GID}
-        data = {'recomm_guid': REC_GID}
-        data['data_date'] = portfolio['date'].strftime('%Y-%m-%d')
-        data['funds'] = [{'weight': round(weight * 100), 'fund_id': id_ticker_map[key]['ftTicker']} for key, weight in
-                         funds.items()]
-        data['creat_date'] = portfolio['create_time'].strftime('%Y-%m-%d %H:%M:%S')
-        # todo 补全
-        # returns = round(datas.pct_change(), 5)
-        # data['cp'] = sharpe_ratio(returns, risk_free=0, period='daily', annualization=None),
-        data['rr'] = 0.81
-        data['roi'] = 0.81
-        data['risk'] = round(sum([id_ticker_map[key]['risk'] * weight for key, weight in funds.items()]), 2)
-        note = {}
-        sig = get_last_signal()
-        note['last_reg_reb'] = sig['date'].strftime('%Y-%m-%d')
+        load_report(min_date=dt.date.today() - dt.timedelta(365))
+        data = {'recomm_guid': REC_GID, 'data_date': sig['date'].strftime('%Y-%m-%d'),
+                'funds': [{'weight': round(weight * 100), 'fund_id': id_ticker_map[key]['ftTicker']} for key, weight in
+                          funds.items()], 'creat_date': sig['create_time'].strftime('%Y-%m-%d %H:%M:%S'),
+                'risk': risk,
+                'rr': round(sum([id_ticker_map[key]['risk'] * weight for key, weight in funds.items()]), 2), 'cp': cp,
+                'roi': roi}
+        note = {'last_rec': next_workday(sig['date']).strftime('%Y%m%d')}
         data['note'] = json.dumps(note)
         portfolios['data'] = data
         rec_list.append(portfolios)
@@ -67,28 +84,29 @@ async def root():
         return {'msg': '当日投组未产生，待10:00后获取'}
 
 
-def start_robo():
+# 其他异常处理程序
+@app.exception_handler(Exception)
+async def general_exception_handler(request: Request, exc: Exception):
+    # 打印一般错误信息
+    logging.error(f"请求 {request.url} 发生未知错误: {str(exc)}")
+    return JSONResponse(
+        status_code=500,
+        content={"errorCode": "500", "errorMsg": str(exc)},
+    )
+
+
+# 定义应用启动事件
+@app.on_event("startup")
+async def startup_event():
     # 异常情况可以重启跑当天投组
     current_time = dt.datetime.now()
     target_time = dt.time(10, 0)
     if current_time.time() > target_time:
-        main.start()
-
+        scheduler.add_job(main.start, trigger=DateTrigger(run_date=current_time))
     # 开启定时任务，执行实盘
-    scheduler = BlockingScheduler()
     scheduler.add_job(main.start, 'cron', day_of_week='0-4', hour=10, minute=00)
     scheduler.start()
 
 
-def start_web():
-    uvicorn.run("robo_controller:app", reload=True, port=8080)
-
-
 if __name__ == "__main__":
-    # 开启一个进程执行start_robo()
-    p1 = Process(target=start_robo)
-    p1.start()
-
-    # 启动进程2
-    p2 = Process(target=start_web)
-    p2.start()
+    uvicorn.run("robo_controller:app", host="0.0.0.0", port=8080)