策略修改

Administrator

2024-07-03 59d0c089c7c0749d121b6fef6cba05509cff6a8c

 trade/backtest_trade.py

@@ -2,40 +2,57 @@
回撤交易
"""
# 持仓字典
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 = {}


# TODO 待完成
def get_price_by_time(code, time_str):
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:
            pass
            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()
        # 回撤
        markets_dict = log_export.load_latest_market_info(date)
        for code in markets_dict:
            __BuyStrategyDataManager.add_market_info(markets_dict[code])
        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"]
            need_buy = __BuyStrategyDataManager.add_transaction_info(t)
            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)

            need_buy, need_buy_msg = __BuyStrategyDataManager.add_transaction_info(t, True)
            if need_buy and code not in position_dict:
                # 持仓结果保存
                position_dict[code] = t
                logger_debug.info(f"回测下单：{t}")
                logger_debug.info(f"回测下单({need_buy_msg})：{t}")
    except Exception as e:
        logging.exception(e)
        logger_debug.exception(e)