import logging import sys from abc import ABC, abstractmethod from datetime import datetime as dt from enum import Enum, unique from py_jftech import get_config, parse_date logger = logging.getLogger(__name__) @unique class DatumType(Enum): FUND = 'FUND' INDEX = 'INDEX' ECO = 'ECO' @unique class PortfoliosRisk(Enum): FT3 = 3 class DataSync(ABC): ''' 数据同步服务,需要同步数据的服务,可以实现该接口 ''' @abstractmethod def do_sync(self, max_date=dt.today()): ''' 开始同步数据,到指定日期,如果没给则同步到当前日期 ''' pass class Datum(ABC): ''' 基础资料服务,基金资料数据,各种指数,指标资料数据 ''' @abstractmethod def get_datums(self, type: DatumType = None, crncy=None, risk=None, datum_ids=None, ticker=None, exclude=True): ''' 获取资料信息,当id和ticker都有时,取二者并集 :param type: 资料类型 :param crncy: 货币类型,仅对基金资料有效 :param risk: 风险等级,仅对基金资料有效 :param datum_ids: 资料ID列表 :param ticker: 资料ticker列表 :param exclude: 是否排除过滤的资料,默认为True :return: 资料信息数据 ''' pass @abstractmethod def get_high_risk_datums(self, risk: PortfoliosRisk): ''' 根据指定的投组风险等级,获取高风险资产资料数据 :param risk: 投组风险等级 :return: 高风险资料信息 ''' pass @abstractmethod def update_change(self, date): ''' 在指定对日期,执行资料变更,如果有变更则返回True, 否则返回False,注意,改方法非幂等,变更后无法复原 :param date: 执行变更的日期 :return: 如果有变更则返回True,否则返回False ''' pass class Navs(ABC): ''' 基础数据相关服务,基金净值,各种指标 高开低收 ''' @abstractmethod def get_fund_navs(self, fund_ids=None, min_date=None, max_date=None): ''' 获取基金净值信息 :param fund_ids: 基金id,可以多个id,使用tuple包裹 :param min_date: 起始时间 :param max_date: 截止时间 :return: 基金净值信息 ''' pass @abstractmethod def get_nav_start_date(self, fund_ids=None): ''' 获取指定id资产的净值开始时间 :param fund_ids: 指定id资产,如果为None,则返回全部资产的开始时间 :return: 资产的开始时间字典 ''' pass @abstractmethod def get_index_close(self, datum_ids=None, min_date=None, max_date=None, ticker=None): ''' 获取指标收盘价 :param datum_ids: 指标资料id :param min_date: 起始时间 :param max_date: 截止时间 :param ticker: 指标资料ticker :return: 指标收盘价信息 ''' pass @abstractmethod def get_last_index_close(self, max_date, datum_id=None, ticker=None, count=1): ''' 获取指定资料或ticker,指定日期之前最后count个收盘价,当指定datum_id后,ticker参数无效 :param max_date: 指定日期 :param datum_id: 指标id,只能指定一个 :param ticker: 指标ticker,只能指定一个,当指标id有值后,该参数无效 :param count: 指定要返回数据的个数 :return: 如果存在,则返回指定日期最后count个收盘价,否则返回None ''' pass @abstractmethod def get_eco_values(self, datum_ids=None, min_date=None, max_date=None, ticker=None, by_release_date=False): ''' 获取经济指标数据,若同时给出ID,和ticker,则取二者并集 :param datum_ids: 经济指标id :param min_date: 起始日期 :param max_date: 截止日期 :param ticker: 经济指标ticker :param by_release_date: 如果为True,则使用公告日期查询,否则使用抓取日期 :return: 经济指标的值,包括查询日期,指标和公告日期 ''' pass @abstractmethod def get_last_eco_values(self, max_date, datum_id=None, ticker=None, count=1, by_release_date=False): ''' 获取指定资料或ticker,指定日期之前最后count个指标数据,当指定datum_id后,ticker参数无效 :param max_date: 指定日期 :param datum_id: 指标id,只能指定一个 :param ticker: 指标ticker,只能指定一个,当指标id有值后,该参数无效 :param count: 指定要返回数据的个数 :param by_release_date: 如果为True,则使用公告日期查询,否则使用抓取日期 :return: 如果存在,则返回指定日期最后count个指标项(查询日期,指标,公告日期),否则返回None ''' pass class RoboExecutor(ABC): ''' ROBO执行器,整合以上逻辑,进行实盘或回测 ''' @abstractmethod def start_exec(self): ''' 开始执行测试逻辑 ''' pass @property @abstractmethod def start_date(self): ''' :return: 执行器开始日期 ''' pass @staticmethod def use_name(): return get_config('robo-executor')['use'] @property def curt_date(self): ''' :return: 当前运行的日期 ''' if len(sys.argv) > 1: try: return parse_date(sys.argv[1]) except Exception as e: logger.warning(f'get curt date from argv failure.', e) return dt.combine(dt.today().date(), dt.min.time())