[Python知识库]量化交易之回测篇 - 通用回测数据源模型（back_tester_source_data.py）

import os
import importlib
import traceback
from datetime import datetime
from pathlib import Path

from server_api.api.tqz_tushare_api import TQZTushareClient
from server_api.api.tqz_tianqin_api import TQZTianQinDataManager
from tqz_strategy.template import CtaTemplate

from public_module.tqz_extern.tools.position_operator.position_operator import TQZJsonOperator
from public_module.tqz_extern.tools.pandas_operator.pandas_operator import pandas

from back_tester_core.back_tester_path import BackTesterDataPath
from back_tester_core.tqz_constant import TQZBackTesterSettingType, TQZStrategyCombinationType, TQZStockPoolExcelType, TQZFuturePoolExcelType
from public_module.tqz_extern.tqz_constant import TQZStockIntervalType


class BackTesterDataCore:

    __back_tester_setting: dict = None

    __strategy_pools_df_map: dict = None
    __bars_map: dict = None

    @classmethod
    def load_strategy_classes(cls, strategy_combination_type: TQZStrategyCombinationType) -> dict:
        """ """
        all_strategies_classes = cls.__get_all_strategy_classes(strategy_combination_type=strategy_combination_type)

        strategies_setting = TQZJsonOperator.tqz_load_jsonfile(
            jsonfile=BackTesterDataPath.get_strategies_setting_path(
                strategy_combination_type=strategy_combination_type
            )
        )

        _strategy_classes = {}
        for value in strategies_setting.values():
            if value["class_name"] in all_strategies_classes.keys() and value["class_name"] not in _strategy_classes.keys() and f'{value["class_name"]}.xlsx' in cls.__get_back_tester_strategy_pools(strategy_combination_type=strategy_combination_type):
                _strategy_classes[value["class_name"]] = {
                    "strategy_class": all_strategies_classes[value["class_name"]],
                    "strategy_setting": value["setting"]
                }

        return _strategy_classes

    @classmethod
    def load_strategyPoolsMap_and_barsMap(cls, strategy_combination_type: TQZStrategyCombinationType) -> (dict, dict):
        """ """
        if cls.__strategy_pools_df_map is None:
            cls.__strategy_pools_df_map = {}
            for dirpath, dirnames, filenames in os.walk(BackTesterDataPath.get_strategy_pools_path(strategy_combination_type=strategy_combination_type)):
                for filename in filenames:
                    if filename.endswith('.xlsx') and filename in cls.__get_back_tester_strategy_pools(strategy_combination_type=strategy_combination_type):
                        if strategy_combination_type in [TQZStrategyCombinationType.STOCK]:
                            cls.__strategy_pools_df_map[filename] = pandas.read_excel(f'{dirpath}/{filename}', sheet_name=TQZStockPoolExcelType.SHEET_NAME.value)
                        elif strategy_combination_type in [TQZStrategyCombinationType.FUTURE]:
                            cls.__strategy_pools_df_map[filename] = pandas.read_excel(f'{dirpath}/{filename}', sheet_name=TQZFuturePoolExcelType.SHEET_NAME.value)
                        else:
                            assert False, f'bad strategy_combination_type: {strategy_combination_type}'

        if cls.__bars_map is None:
            if strategy_combination_type in [TQZStrategyCombinationType.STOCK]:
                back_tester_setting = cls.__get_back_tester_setting(strategy_combination_type=strategy_combination_type)

                merge_stocks = []
                for stock_pool_dataframe in cls.__strategy_pools_df_map.values():
                    merge_stocks += stock_pool_dataframe[TQZStockPoolExcelType.STOCKS_COLUMN.value].values.tolist()

                cls.__bars_map = TQZTushareClient.query_multi_stocks_history_bars(
                    stock_list=list(set(merge_stocks)),
                    start=datetime.strptime(back_tester_setting[TQZBackTesterSettingType.START_DATE.value], '%Y-%m-%d'),
                    end=datetime.strptime(back_tester_setting[TQZBackTesterSettingType.END_DATE.value], '%Y-%m-%d'),
                    interval=cls.__get_current_interval()
                )
            elif strategy_combination_type in [TQZStrategyCombinationType.FUTURE]:
                back_tester_setting = cls.__get_back_tester_setting(strategy_combination_type=strategy_combination_type)

                future_contracts = []
                for future_pool_dataframe in cls.__strategy_pools_df_map.values():
                    future_contracts += future_pool_dataframe[TQZFuturePoolExcelType.CONTRACTS_COLUMN.value].values.tolist()

                cls.__bars_map = cls.__get_future_bars_map(future_contracts=future_contracts, back_tester_setting=back_tester_setting)
            else:
                assert False, f'bad strategy_combination_type: {strategy_combination_type}'

        return cls.__strategy_pools_df_map, cls.__bars_map


    @classmethod
    def get_slippage(cls, strategy_combination_type: TQZStrategyCombinationType):
        return cls.__get_back_tester_setting(
            strategy_combination_type=strategy_combination_type
        )[TQZBackTesterSettingType.SLIPPAGE.value]

    @classmethod
    def get_fee_ratio(cls, strategy_combination_type: TQZStrategyCombinationType):
        return cls.__get_back_tester_setting(
            strategy_combination_type=strategy_combination_type
        )[TQZBackTesterSettingType.FEE_RATIO.value]

    @classmethod
    def get_stock_tick_value(cls):
        return cls.__get_back_tester_setting(
            strategy_combination_type=TQZStrategyCombinationType.STOCK
        )[TQZBackTesterSettingType.TICK_VALUE.value]

    @classmethod
    def get_future_contracts_setting(cls):
        return TQZJsonOperator.tqz_load_jsonfile(jsonfile=BackTesterDataPath.get_future_contracts_setting_path())


    # --- private part ---
    @classmethod
    def __get_future_bars_map(cls, future_contracts: list, back_tester_setting: dict) -> dict:
        contracts, main_contracts, index_contracts = [], [], []
        for contract in future_contracts:
            if 'i@' in contract:
                index_contracts.append(contract)
            elif 'm@' in contract:
                main_contracts.append(contract)
            else:
                contracts.append(contract)

        contract_bars_map = TQZTianQinDataManager.load_history_bars_map(
            tq_symbols=list(set(contracts)),
            tq_duration_seconds=back_tester_setting[TQZBackTesterSettingType.BACK_TESTER_INTERVAL_SECONDES.value],
            tq_data_length=3
            # tq_data_length=back_tester_setting[TQZBackTesterSettingType.BACK_TESTER_BARS_LENGTH.value]
        )
        main_contract_bars_map = TQZTianQinDataManager.load_history_main_bars_map(
            tq_main_symbols=list(set(main_contracts)),
            tq_duration_seconds=back_tester_setting[TQZBackTesterSettingType.BACK_TESTER_INTERVAL_SECONDES.value],
            tq_data_length=3
            # tq_data_length = back_tester_setting[TQZBackTesterSettingType.BACK_TESTER_BARS_LENGTH.value]
        )
        index_contract_bars_map = TQZTianQinDataManager.load_history_index_bars_map(
            tq_index_symbols=list(set(index_contracts)),
            tq_duration_seconds=back_tester_setting[TQZBackTesterSettingType.BACK_TESTER_INTERVAL_SECONDES.value],
            tq_data_length=3
            # tq_data_length = back_tester_setting[TQZBackTesterSettingType.BACK_TESTER_BARS_LENGTH.value]
        )

        return {**contract_bars_map, **main_contract_bars_map, **index_contract_bars_map}

    @classmethod
    def __get_back_tester_strategy_pools(cls, strategy_combination_type: TQZStrategyCombinationType):
        if strategy_combination_type in [TQZStrategyCombinationType.STOCK]:
            return cls.__get_back_tester_setting(
                strategy_combination_type=strategy_combination_type
            )[TQZBackTesterSettingType.BACK_TESTER_STOCK_POOLS.value]
        elif strategy_combination_type in [TQZStrategyCombinationType.FUTURE]:
            return cls.__get_back_tester_setting(
                strategy_combination_type=strategy_combination_type
            )[TQZBackTesterSettingType.BACK_TESTER_FUTURE_POOLS.value]
        else:
            assert False, f'bad strategy_combination_type: {strategy_combination_type}'

    @classmethod
    def __get_current_interval(cls) -> TQZStockIntervalType:
        interval_str = cls.__get_back_tester_setting(strategy_combination_type=TQZStrategyCombinationType.STOCK)[TQZBackTesterSettingType.INTERVAL.value]

        if interval_str in [TQZStockIntervalType.DAILY.value]:
            interval = TQZStockIntervalType.DAILY
        elif interval_str in [TQZStockIntervalType.WEEKLY.value]:
            interval = TQZStockIntervalType.WEEKLY
        elif interval_str in [TQZStockIntervalType.MONTHLY.value]:
            interval = TQZStockIntervalType.MONTHLY
        else:
            assert False, f'bad interval_str {interval_str}'

        return interval

    @classmethod
    def __get_back_tester_setting(cls, strategy_combination_type: TQZStrategyCombinationType):
        if cls.__back_tester_setting is None:
            cls.__back_tester_setting = TQZJsonOperator.tqz_load_jsonfile(
                jsonfile=BackTesterDataPath.get_back_tester_setting_path(strategy_combination_type=strategy_combination_type)
            )

        return cls.__back_tester_setting

    @classmethod
    def __get_all_strategy_classes(cls, strategy_combination_type: TQZStrategyCombinationType) -> dict:
        """ """
        root = Path(__file__).parent.parent
        stock_strategies_path = root.joinpath("tqz_strategy", f'{strategy_combination_type.value}', "strategies")
        module_name = f'tqz_strategy.{strategy_combination_type.value}.strategies'

        strategies_classes = {}

        for dirpath, dirnames, filenames in os.walk(stock_strategies_path):
            for filename in filenames:
                if filename.endswith(".py"):
                    strategy_module_name = ".".join(
                        [module_name, filename.replace(".py", "")])

                    try:
                        module = importlib.import_module(strategy_module_name)
                        importlib.reload(module)

                        for name in dir(module):
                            value = getattr(module, name)
                            if isinstance(value, type) and issubclass(value, CtaTemplate) and value is not CtaTemplate:
                                strategies_classes[value.__name__] = value
                    except:  # noqa
                        assert False, f"策略文件{module_name}加载失败，触发异常：\n{traceback.format_exc()}"

        return strategies_classes


if __name__ == '__main__':
    """
    stock_pools_map, history_bars_map = BackTesterDataCore.load_strategyPoolsMap_and_barsMap(strategy_combination_type=TQZStrategyCombinationType.STOCK)
    strategy_classes = BackTesterDataCore.load_strategy_classes(strategy_combination_type=TQZStrategyCombinationType.STOCK)  # strategy_name strategy_classes

    print("stock_pools_map: " + str(stock_pools_map))
    print("history_bars_map: " + str(history_bars_map))
    print("strategy_classes: " + str(strategy_classes))
    """

    future_pools_map, history_bars_map = BackTesterDataCore.load_strategyPoolsMap_and_barsMap(strategy_combination_type=TQZStrategyCombinationType.FUTURE)
    strategy_classes = BackTesterDataCore.load_strategy_classes(strategy_combination_type=TQZStrategyCombinationType.FUTURE)  # strategy_name strategy_classes

    print("future_pools_map: " + str(future_pools_map))
    print("history_bars_map: " + str(history_bars_map))
    print("strategy_classes: " + str(strategy_classes))