import re
from public_module.tqz_extern.tools.pandas_operator.pandas_operator import pandas
from public_module.tqz_extern.tools.position_operator.position_operator import TQZJsonOperator
from public_module.object import BarData
from back_tester_branch.back_tester_source_data import TQZBackTesterFutureSourceData
from back_tester_source_data import TQZBackTesterFutureConfigPath
class TQZBackTesterFuture:
__future_contracts_setting = TQZJsonOperator.tqz_load_jsonfile(
jsonfile=TQZBackTesterFutureConfigPath.future_contracts_setting_path()
)
@classmethod
def start(cls):
_strategies = TQZBackTesterFutureSourceData.load_future_strategies()
back_tester_result = {}
for strategy_name, data in _strategies.items():
strategy, bars = data['strategy'], data['bars']
per_result_df, daily_per_result_df = pandas.DataFrame(columns={'date', 'close_price', 'pos'}), pandas.DataFrame(columns={'date', 'balance'})
# back tester single strategy
strategy.on_init()
strategy.on_start()
strategy_fund = cls.__get_contract_fund(tq_sym=bars[0].vt_symbol)
for bar in bars:
strategy.on_bar(bar)
per_result_df = cls.__update_per_result_df(per_result_df=per_result_df, bar=bar, strategy=strategy)
search_result = re.search('(\d+)-(\d+)-(\d+) 14:(\d+):(\d+)', bar.datetime)
if search_result is not None:
daily_per_result_df_current_row = len(daily_per_result_df)
daily_per_result_df.loc[daily_per_result_df_current_row, 'date'] = re.match('(\d+)-(\d+)-(\d+)', bar.datetime).group()
daily_per_result_df.loc[daily_per_result_df_current_row, 'balance'] = per_result_df[-1:]['balance'].tolist()[0]
daily_per_result_df.loc[daily_per_result_df_current_row, 'net_value'] = daily_per_result_df.loc[daily_per_result_df_current_row, 'balance'] / strategy_fund
daily_per_result_df['fund'] = strategy_fund
strategy.on_stop()
back_tester_result[strategy_name] = daily_per_result_df
cls.__settle_back_tester_result(back_tester_result=back_tester_result)
# --- private part ---
@classmethod
def __settle_back_tester_result(cls, back_tester_result: {str, pandas.DataFrame}):
pass
@classmethod
def __update_per_result_df(cls, per_result_df: pandas.DataFrame, bar: BarData, strategy) -> pandas.DataFrame:
current_row = len(per_result_df)
per_result_df.loc[current_row, 'date'] = bar.datetime
per_result_df.loc[current_row, 'close_price'] = bar.close_price
per_result_df.loc[current_row, 'pos'] = strategy.pos
if current_row is 0:
per_result_df.loc[current_row, 'cc_diff'] = 0
else:
per_result_df.loc[current_row, 'cc_diff'] = bar.close_price - per_result_df.loc[current_row - 1, 'close_price']
per_result_df.loc[current_row, 'cc_pnl'] = per_result_df.loc[current_row, 'cc_diff'] * per_result_df.loc[current_row, 'pos']
if current_row is 0:
per_result_df.loc[current_row, 'pnl'] = 0
elif current_row is not 0 and per_result_df.loc[current_row, 'pos'] is not 0 and per_result_df.loc[current_row - 1, 'pos'] is 0:
per_result_df.loc[current_row, 'pnl'] = per_result_df.loc[current_row - 1, 'pnl']
else:
per_result_df.loc[current_row, 'pnl'] = per_result_df.loc[current_row, 'cc_pnl'] + per_result_df.loc[current_row - 1, 'pnl']
per_result_df.loc[current_row, 'real_pnl'] = per_result_df.loc[current_row, 'pnl'] * cls.__future_contracts_setting[strategy.vt_symbol]['contract_multiple']
per_result_df.loc[current_row, 'balance'] = per_result_df.loc[current_row, 'real_pnl'] + cls.__get_contract_fund(tq_sym=strategy.vt_symbol)
return per_result_df
@classmethod
def __get_contract_fund(cls, tq_sym: str) -> float:
future_contracts_fund = {
"CFFEX.T": 100000,
"CFFEX.IF": 1000000,
"CFFEX.IC": 1000000,
"CFFEX.IH": 1000000
}
if tq_sym in future_contracts_fund.keys():
return future_contracts_fund[tq_sym]
else:
return 50000
if __name__ == '__main__':
TQZBackTesterFuture.start()
|
