| | |
| | | """ |
| | | # 持仓字典 |
| | | from log_module import log_export |
| | | from log_module.log import logger_debug |
| | | from trade.buy_strategy import BuyStrategyDataManager |
| | | |
| | | position_dict = {} |
| | | |
| | | |
| | | def start_backtest(date): |
| | | __BuyStrategyDataManager = BuyStrategyDataManager() |
| | | # 回撤 |
| | | markets_dict = log_export.load_latest_market_info(date) |
| | | for code in markets_dict: |
| | | __BuyStrategyDataManager.add_market_info(markets_dict[code]) |
| | | transactions = log_export.load_transactions(date) |
| | | for t in transactions: |
| | | code = t["SecurityID"] |
| | | need_buy = __BuyStrategyDataManager.add_transaction_info(t) |
| | | if need_buy and code not in position_dict: |
| | | # 持仓结果保存 |
| | | position_dict[code] = t |
| | | try: |
| | | __BuyStrategyDataManager = BuyStrategyDataManager() |
| | | # 回撤 |
| | | markets_dict = log_export.load_latest_market_info(date) |
| | | for code in markets_dict: |
| | | __BuyStrategyDataManager.add_market_info(markets_dict[code]) |
| | | transactions = log_export.load_transactions(date) |
| | | for t in transactions: |
| | | code = t["SecurityID"] |
| | | need_buy = __BuyStrategyDataManager.add_transaction_info(t) |
| | | if need_buy and code not in position_dict: |
| | | # 持仓结果保存 |
| | | position_dict[code] = t |
| | | logger_debug.info(f"回测下单:{t}") |
| | | except Exception as e: |
| | | logger_debug.exception(e) |