from datetime import datetime as dt from typing import List import pandas as pd from py_jftech import component, autowired, workday_range, parse_date from api import AssetPool, AssetOptimize, AssetRisk, RoboReportor, Datum, DatumType, RoboExecutor from asset_pool.dao import robo_assets_pool as rap @component class FundAssetPool(AssetPool): @autowired def __init__(self, optimize: AssetOptimize = None, risk: AssetRisk = None): self._optimize = optimize self._risk = risk def get_pool(self, day=dt.today()): opti_pool = self._optimize.get_optimize_pool(day) risk_pool = self._risk.get_risk_pool(day) return [x for x in opti_pool if x not in risk_pool] def clear(self, day=None): rap.delete(day) @component(bean_name='asset-pool-report') class AssetPoolReportor(RoboReportor): @autowired def __init__(self, optimize: AssetOptimize = None, risk: AssetRisk = None, datum: Datum = None, executor: RoboExecutor = None): self._optimize = optimize self._risk = risk self._datum = datum self._executor = executor @property def report_name(self) -> str: return '基金池' def load_report(self, max_date=dt.today(), min_date=None) -> List[dict]: datums = self._datum.get_datums(type=DatumType.FUND) result = [] for date in workday_range(min_date or self._executor.start_date, max_date): opts = self._optimize.get_optimize_pool(date) risk = self._risk.get_risk_pool(date) datas = {x['bloombergTicker']: 0 if x['id'] in risk else 1 if x['id'] in opts else -1 for x in datums} result.append({'date': date, **datas}) return result