"""
|
回撤交易
|
"""
|
# 持仓字典
|
import logging
|
|
from log_module import log_export
|
from log_module.log import logger_debug
|
from trade.buy_strategy import BuyStrategyDataManager
|
from utils import l2_huaxin_util, tool
|
|
position_dict = {}
|
|
markets_list_dict = {}
|
|
|
def get_market_by_time(code, time_str):
|
"""
|
获取某个时间的价格
|
:param code:
|
:param time_str:
|
:return:
|
"""
|
markets = markets_list_dict.get(code)
|
if markets:
|
for m in markets:
|
if tool.trade_time_sub(l2_huaxin_util.convert_time(m[9]), time_str) >= 0:
|
return m
|
return None
|
|
|
def start_backtest(date):
|
"""
|
开始回测
|
:param date:
|
:return:
|
"""
|
try:
|
__BuyStrategyDataManager = BuyStrategyDataManager()
|
# 回撤
|
global markets_list_dict
|
markets_list_dict = log_export.load_market_info(date)
|
transactions = log_export.load_transactions(date)
|
for t in transactions:
|
code = t["SecurityID"]
|
time_str = l2_huaxin_util.convert_time(t["OrderTime"])
|
market_info = get_market_by_time(code, time_str)
|
if market_info:
|
__BuyStrategyDataManager.add_market_info(market_info)
|
|
results = __BuyStrategyDataManager.add_transaction_info(t, True)
|
for need_buy, strategy_type, need_buy_msg in results:
|
if need_buy and code not in position_dict:
|
# 持仓结果保存
|
position_dict[code] = t
|
logger_debug.info(f"回测下单({need_buy_msg})：{t}")
|
except Exception as e:
|
logging.exception(e)
|
logger_debug.exception(e)
|
