
等级: 新手上路
- 注册:
- 2023-9-8
- 曾用名:
|
`python
import backtrader as bt
import pandas as pd
class DualMAStrategy(bt.Strategy):
params = (
('fast_period', 10), # 快速均线周期
('slow_period', 30), # 慢速均线周期
('risk_percent', 1), # 单笔风险比例
('atr_period', 14), # ATR波动率计算周期
('rsi_period', 14), # RSI过滤条件
('trading_cost', 0.0002), # 交易成本
)
def __init__(self):
# 均线交叉系统
self.fast_ma = bt.indicators.SMA(period=self.p.fast_period)
self.slow_ma = bt.indicators.SMA(period=self.p.slow_period)
self.crossover = bt.indicators.CrossOver(self.fast_ma, self.slow_ma)
# 风险控制指标
self.atr = bt.indicators.ATR(period=self.p.atr_period)
self.rsi = bt.indicators.RSI(period=self.p.rsi_period)
self.order = None
def next(self):
if self.order: # 检查是否有挂单
return
# 头寸管理
position_size = self.broker.getvalue() * self.p.risk_percent / 100 / self.atr[0]
# 交易逻辑
if not self.position: # 没有持仓时
if self.crossover > 0 and self.rsi < 70: # 金叉且不过度超买
self.order = self.buy(size=position_size)
elif self.crossover < 0 and self.rsi > 30: # 死叉且不过度超卖
self.order = self.sell(size=position_size)
else: # 持仓时
# 动态止盈止损(2:1 盈亏比)
if self.position.size > 0: # 多头持仓
if self.data.close[0] >= self.entry_price + 2 * self.atr[0]:
self.close()
elif self.data.close[0] <= self.entry_price - self.atr[0]:
self.close()
else: # 空头持仓
if self.data.close[0] <= self.entry_price - 2 * self.atr[0]:
self.close()
elif self.data.close[0] >= self.entry_price + self.atr[0]:
self.close()
def notify_order(self, order):
if order.status in [order.Completed]:
self.entry_price = order.executed.price # 记录入场价格
self.order = None
if __name__ == '__main__':
cerebro = bt.Cerebro()
# 数据加载(示例用螺纹钢期货)
data = pd.read_csv('RB8888.csv', parse_dates=['datetime'], index_col='datetime')
datafeed = bt.feeds.PandasData(dataname=data)
cerebro.adddata(datafeed)
# 策略参数优化空间
cerebro.optstrategy(
DualMAStrategy,
fast_period=range(5, 20, 5),
slow_period=range(20, 50, 10),
risk_percent=[0.5, 1, 2]
)
# 初始资金10万元
cerebro.broker.setcash(100000.0)
# 设置交易成本(双边)
cerebro.broker.setcommission(commission=self.p.trading_cost)
# 运行回测
results = cerebro.run()
# 绩效分析
cerebro.addanalyzer(bt.analyzers.PyFolio, _name='pyfolio')
strategies = results[0]
pyfoliozer = strategies.analyzers.getbyname('pyfolio')
returns, positions, transactions, gross_lev = pyfoliozer.get_pf_items()
# 输出关键指标
print(f'最终资金: {cerebro.broker.getvalue():.2f}')
print(f'年化收益率: {annual_return*100:.2f}%')
print(f'最大回撤: {max_drawdown*100:.2f}%')
print(f'胜率: {win_rate*100:.2f}%')
print(f'盈亏比: {profit_ratio:.2f}')
```
|
|