feat(execution): 实现执行层(回测 + Dry-run)

核心组件:
- Executor: 执行器抽象基类
- BacktestExecutor: 回测执行器
  - 处理信号、计算净值、记录交易
  - 支持交易成本设置
- DryRunExecutor: 模拟盘执行器
  - 模拟下单、模拟成交、模拟持仓更新
  - 不影响真实资金
- Portfolio: 持仓组合数据类

特点:
- 统一接口(execute方法)
- 支持两种模式切换(回测/Dry-run)
- 实盘执行器预留扩展点

测试覆盖:7个测试全部通过
This commit is contained in:
2026-05-11 22:19:07 +08:00
parent 7468130450
commit babf224203
2 changed files with 280 additions and 0 deletions

View File

@@ -0,0 +1,102 @@
"""
执行层测试
"""
import pytest
import pandas as pd
import numpy as np
from framework.execution import Executor, BacktestExecutor, DryRunExecutor, Portfolio
class TestExecutor:
"""测试执行器基类"""
def test_executor_mode(self):
"""测试执行器模式"""
backtest = BacktestExecutor()
assert backtest.get_mode() == "backtest"
dry_run = DryRunExecutor()
assert dry_run.get_mode() == "dry_run"
class TestBacktestExecutor:
"""测试回测执行器"""
def test_backtest_init(self):
"""测试回测初始化"""
executor = BacktestExecutor(
initial_capital=100000.0,
trade_cost=0.001
)
assert executor.initial_capital == 100000.0
assert executor.trade_cost == 0.001
def test_backtest_execute(self):
"""测试回测执行"""
executor = BacktestExecutor(initial_capital=100000.0)
# 创建测试数据
dates = pd.date_range('2020-01-01', periods=10)
signals = pd.DataFrame({
'signal': ['code1,code2'] * 10
}, index=dates)
data = pd.DataFrame({
'code1': [100.0] * 10,
'code2': [50.0] * 10,
}, index=dates)
portfolio = executor.execute(signals, data)
assert portfolio is not None
assert portfolio.cash == 100000.0
class TestDryRunExecutor:
"""测试模拟盘执行器"""
def test_dry_run_init(self):
"""测试模拟盘初始化"""
executor = DryRunExecutor(initial_capital=50000.0)
assert executor.initial_capital == 50000.0
def test_simulate_order(self):
"""测试模拟下单"""
executor = DryRunExecutor(initial_capital=100000.0)
# 初始化持仓
executor._portfolio = Portfolio(
positions={},
cash=100000.0,
nav=1.0,
trades=[]
)
# 模拟买入
executor.simulate_order('code1', 'BUY', 100, 50.0)
# 检查现金减少
assert executor._portfolio.cash == 100000.0 - 100 * 50.0
class TestPortfolio:
"""测试持仓组合"""
def test_portfolio_value(self):
"""测试持仓价值计算"""
portfolio = Portfolio(
positions={},
cash=50000.0,
nav=1.0,
trades=[]
)
assert portfolio.get_total_value() == 50000.0
if __name__ == '__main__':
pytest.main([__file__, '-v'])