量化交易之One Piece篇 - back tester 新增 PluginOperator

import re
import os
import pandas
from op_futures.op_strategy_plugins.template_strategy import TemplateStrategy
from op_futures.op_backtester.op_generate_bar import BarGenerate
from op_futures.op_objects.onepiece_data import OnePieceData
from op_futures.op_objects.tick_data import TickData
from op_futures.op_objects.bar_data import BarData

from tqz_extern.local_database import LocalDB
from tqz_extern.json_operator import TQZJsonOperator
from tqz_extern.tqz_constant import BackTesterType, OrderSide


class PluginOperator:

    @classmethod
    def make_plugins_config(cls, backtester_datetime_list: list) -> dict:
        strategies_template_config = TQZJsonOperator.tqz_load_jsonfile(
            jsonfile='strategies_template_config.json'
        )

        plugins_config = {}
        for backtester_datetime in backtester_datetime_list:
            main_contracts = cls.__get_main_contracts_list(backtester_datetime=backtester_datetime)
            for main_contract in main_contracts:
                for strategy_name, strategy_params in strategies_template_config.items():
                    # datas path.
                    single_main_contract_path = f'{LocalDB.market_depth_format_data_dir(datetime_str=backtester_datetime)}/{main_contract}.csv'
                    assert os.path.exists(path=single_main_contract_path), f'{single_main_contract_path} is exist.'

                    # plugin_name.
                    plugin_name = cls.__get_plugin_name(main_contract=main_contract, strategy_name=strategy_name, strategy_params=strategy_params)

                    plugin_key = f'{plugin_name}_{backtester_datetime}'
                    if plugin_key in plugins_config.keys():
                        assert False, f'same {plugin_key} in plugins_config: {str(plugins_config)}'

                    plugins_config[plugin_key] = {
                        'strategy_name': str(strategy_name),
                        'plugin_name': str(plugin_name),
                        'symbol': str(main_contract),
                        'date_time': str(backtester_datetime),
                        'single_main_contract_path': single_main_contract_path,
                        'interval_secs': strategy_params["interval_secs"],
                        'params': strategy_params["params"]
                    }

        return plugins_config

    @classmethod
    def make_plugins(cls, plugins_config: dict, strategy_classes: [TemplateStrategy]):
        plugins = []
        for plugin_name, plugin_config in plugins_config.items():
            strategy = strategy_classes[plugin_config['strategy_name']]
            if 0 == plugin_config['interval_secs']:
                datas = BarGenerate(plugin_config['single_main_contract_path']).ticks_list_value()
                back_tester_type = BackTesterType.TICK_TYPE
            else:
                datas = BarGenerate(plugin_config['single_main_contract_path']).make_bar_dataframe(interval_secs=plugin_config['interval_secs']).bars_list_value()
                back_tester_type = BackTesterType.BAR_TYPE

            plugins.append(
                PluginData(
                    name=plugin_config['plugin_name'],
                    date_time=plugin_config['date_time'],
                    strategy=strategy(params=plugin_config['params']),
                    datas=datas,
                    back_tester_type=back_tester_type
                )
            )

        return plugins

    @classmethod
    def __get_main_contracts_list(cls, backtester_datetime) -> [str]:
        datas_csv_path = f'{LocalDB.daily_main_contracts_config_dir()}/{backtester_datetime}_main_contracts.csv'
        assert os.path.exists(path=datas_csv_path), f'path is exist.'

        main_contracts = []
        [main_contracts.append(main_contract.split('.')[1]) for main_contract in pandas.read_csv(datas_csv_path)['ExchangeInstrument'].tolist()]

        return main_contracts

    @classmethod
    def __get_plugin_name(cls, main_contract: str, strategy_name: str, strategy_params: dict) -> str:
        main_symbol = re.sub(r'\d+', '', main_contract)
        plugin_name = f'{strategy_name}_{main_symbol}888'

        if 'DoubleMaStrategy' == strategy_name:  # <- 'DoubleMaStrategy'
            assert 'interval_secs' in strategy_params.keys(), f'strategies_template_config.json {strategy_name} has no interval_secs.'
            plugin_name += f'_interval_secs_{strategy_params["interval_secs"]}'
            assert 'params' in strategy_params.keys(), f'strategies_template_config.json {strategy_name} has no params.'
            assert 'fast_window' in strategy_params[
                'params'].keys(), f'strategies_template_config.json {strategy_name} has no fast_window.'
            plugin_name += f'_fast_window_{strategy_params["params"]["fast_window"]}'
            assert 'slow_window' in strategy_params[
                'params'].keys(), f'strategies_template_config.json {strategy_name} has no slow_window.'
            plugin_name += f'_slow_window_{strategy_params["params"]["slow_window"]}'
        else:
            assert False, f'Bad strategy: {strategy_name}'

        return plugin_name