import pandas
import os
from datetime import datetime
from op_futures.op_objects.plugin_data import PluginData, PluginOperator
from tqz_extern.tqz_constant import BackTesterType, StrategyMarket
from tqz_extern.local_database import LocalDB
from tqz_extern.strategy_classes import StrategyClasses
class BackTesterEngine:
def __init__(self, plugins: [PluginData]):
self.__plugins: [PluginData] = plugins
def run(self, dump_orders_detail_csv: bool = False, slippage_counts: int = 1):
"""
start back tester engine.
:param dump_orders_detail_csv: dump orders_detail.csv or not.
:param slippage_counts: slippage counts.
:return:
"""
self.__check_back_tester() # <- check all plugin is run able.
print(f'[{datetime.now()} onepiece_rsh]: back tester engine start.')
for plugin_data in self.__plugins: # do all plugins back testing.
plugin_data.strategy.on_init()
if plugin_data.back_tester_type is BackTesterType.BAR_TYPE:
[plugin_data.strategy.on_bar(bar=bar_data) for bar_data in plugin_data.datas]
elif plugin_data.back_tester_type is BackTesterType.TICK_TYPE:
[plugin_data.strategy.on_tick(tick=tick_data) for tick_data in plugin_data.datas]
else:
assert True, f'Bad back_tester_type: {plugin_data.back_tester_type}.'
plugin_data.strategy.on_stop()
if dump_orders_detail_csv: # <- dump orders_detail_csv;
self.__make_orders_detail_df(plugin_data=plugin_data).to_csv(
f'{LocalDB.back_tester_result_orders_details_dir()}/{plugin_data.name}_orders_detail_{plugin_data.date_time}.csv',
index=False
)
# calculate profit_and_loss & slippage inside.
plugin_data.back_testing(slippage_counts=slippage_counts)
print(f'[{datetime.now()} onepiece_rsh]: back tester engine end.')
return self
def dump_back_tester_result(self):
for plugin_data in self.__plugins:
single_symbol_back_tester_csv = f'{LocalDB.back_tester_result_dir()}/{plugin_data.name}.csv'
if os.path.exists(path=single_symbol_back_tester_csv) is False:
last_line_map = {
'Date': [plugin_data.date_time],
'profit_and_loss_value': [str(plugin_data.profit_and_loss_value())],
'slippage_loss_money': [str(plugin_data.slippage_loss_money())],
}
single_symbol_back_tester_df = pandas.DataFrame(last_line_map)
single_symbol_back_tester_df.to_csv(single_symbol_back_tester_csv, index=False)
else:
single_symbol_back_tester_df = pandas.read_csv(single_symbol_back_tester_csv)
last_line_map = {
'Date': plugin_data.date_time,
'profit_and_loss_value': str(plugin_data.profit_and_loss_value()),
'slippage_loss_money': str(plugin_data.slippage_loss_money()),
}
if last_line_map['Date'] in single_symbol_back_tester_df['Date'].astype(str).values.tolist():
single_symbol_back_tester_df = single_symbol_back_tester_df[single_symbol_back_tester_df['Date'].astype(str) != last_line_map['Date']]
single_symbol_back_tester_df = single_symbol_back_tester_df.append(last_line_map, ignore_index=True)
single_symbol_back_tester_df['Date'] = single_symbol_back_tester_df['Date'].astype(str)
single_symbol_back_tester_df.sort_values(by='Date', ascending=True, inplace=True)
single_symbol_back_tester_df.reset_index(inplace=True)
del single_symbol_back_tester_df['index']
single_symbol_back_tester_df.to_csv(single_symbol_back_tester_csv, index=False)
def __check_back_tester(self) -> bool:
assert len(self.__plugins) > 0, f'strategy plugins is empty.'
for plugin_data in self.__plugins:
assert len(plugin_data.datas) > 0, f'plugin {plugin_data.name} history data is empty.'
if os.path.exists(path=LocalDB.back_tester_result_dir()) is False:
os.makedirs(LocalDB.back_tester_result_dir(), exist_ok=True)
if os.path.exists(path=LocalDB.back_tester_result_orders_details_dir()) is False:
os.makedirs(LocalDB.back_tester_result_orders_details_dir(), exist_ok=True)
@staticmethod
def __make_orders_detail_df(plugin_data):
orders_detail_df = pandas.DataFrame(
columns=['symbol', 'date_time', 'lots', 'order_side', 'order_type', 'price']
)
for order in plugin_data.strategy.orders_detail:
orders_detail_df.loc[len(orders_detail_df)] = {
'symbol': str(order.symbol),
'date_time': order.date_time,
'lots': str(order.lots),
'order_side': str(order.order_side),
'order_type': str(order.order_type),
'price': str(order.price)
}
if hasattr(plugin_data.datas[-1], 'close_price'):
orders_detail_df['daily_last_price'] = plugin_data.datas[-1].close_price
elif hasattr(plugin_data.datas[-1], 'last_price'):
orders_detail_df['daily_last_price'] = plugin_data.datas[-1].last_price
return orders_detail_df
if __name__ == '__main__':
BackTesterEngine(
plugins=PluginOperator.make_plugins(
plugins_config=PluginOperator.make_plugins_config(
backtester_datetime_list=['20230926', '20230928']
),
strategy_classes=StrategyClasses.load_config(
strategy_market=StrategyMarket.FUTURES
)
)
).run(dump_orders_detail_csv=True).dump_back_tester_result()