实战项目：完整的趋势跟踪策略系统
wabicai
# 实战项目：完整的趋势跟踪策略系统
# 学习目标
通过本项目,你将学会:
- 从零构建一个完整的趋势跟踪系统
- 实现多种趋势识别算法
- 加入风险管理模块
- 进行全面的回测和优化
- 生成专业的分析报告
# 项目概述
# 系统架构
数据层(Data Layer)
↓
指标层(Indicator Layer)
↓
信号层(Signal Layer)
↓
风险层(Risk Management Layer)
↓
执行层(Execution Layer)
↓
监控层(Monitor Layer)
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
# 功能特性
多指标趋势识别
- 移动平均线系统
- ADX趋势强度
- 唐奇安通道突破
风险管理
- 动态止损
- 仓位管理
- 最大回撤控制
回测分析
- 多维度性能指标
- 可视化报告
- 参数优化
# 完整代码实现
"""
趋势跟踪策略完整系统
作者: Your Name
日期: 2024-01-01
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import yfinance as yf
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Dict, List, Tuple
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False
@dataclass
class TradeSignal:
"""交易信号"""
date: datetime
symbol: str
signal_type: str # 'BUY' or 'SELL'
price: float
reason: str
@dataclass
class Position:
"""持仓信息"""
symbol: str
entry_date: datetime
entry_price: float
quantity: int
stop_loss: float = None
take_profit: float = None
class TrendFollowingSystem:
"""趋势跟踪策略系统"""
def __init__(self, config: Dict):
"""
初始化系统
:param config: 配置字典
- symbol: 股票代码
- start_date: 开始日期
- end_date: 结束日期
- initial_capital: 初始资金
- fast_ma: 快线周期
- slow_ma: 慢线周期
- adx_period: ADX周期
- adx_threshold: ADX阈值
- atr_period: ATR周期
- atr_multiplier: ATR倍数(止损)
- position_size: 仓位大小
"""
self.config = config
self.data = None
self.signals = []
self.trades = []
self.positions = {}
self.equity_curve = []
self.cash = config['initial_capital']
self.total_value = config['initial_capital']
def download_data(self):
"""下载数据"""
print(f"下载 {self.config['symbol']} 数据...")
self.data = yf.download(
self.config['symbol'],
start=self.config['start_date'],
end=self.config['end_date']
)
print(f"下载完成！共 {len(self.data)} 条记录\n")
return self
def calculate_indicators(self):
"""计算技术指标"""
print("计算技术指标...")
# 移动平均线
self.data['fast_ma'] = self.data['Close'].rolling(
window=self.config['fast_ma']
).mean()
self.data['slow_ma'] = self.data['Close'].rolling(
window=self.config['slow_ma']
).mean()
# ATR(平均真实波幅)
self.data['atr'] = self._calculate_atr(self.config['atr_period'])
# ADX(趋势强度指标)
self.data['adx'] = self._calculate_adx(self.config['adx_period'])
# 唐奇安通道
channel_period = 20
self.data['upper_channel'] = self.data['High'].rolling(channel_period).max()
self.data['lower_channel'] = self.data['Low'].rolling(channel_period).min()
print("指标计算完成!\n")
return self
def _calculate_atr(self, period: int) -> pd.Series:
"""计算ATR"""
high_low = self.data['High'] - self.data['Low']
high_close = np.abs(self.data['High'] - self.data['Close'].shift())
low_close = np.abs(self.data['Low'] - self.data['Close'].shift())
ranges = pd.concat([high_low, high_close, low_close], axis=1)
true_range = ranges.max(axis=1)
atr = true_range.rolling(period).mean()
return atr
def _calculate_adx(self, period: int) -> pd.Series:
"""计算ADX"""
# 计算+DM和-DM
high_diff = self.data['High'].diff()
low_diff = -self.data['Low'].diff()
plus_dm = high_diff.where((high_diff > low_diff) & (high_diff > 0), 0)
minus_dm = low_diff.where((low_diff > high_diff) & (low_diff > 0), 0)
# 计算ATR
atr = self._calculate_atr(period)
# 计算+DI和-DI
plus_di = 100 * (plus_dm.rolling(period).mean() / atr)
minus_di = 100 * (minus_dm.rolling(period).mean() / atr)
# 计算DX和ADX
dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di)
adx = dx.rolling(period).mean()
return adx
def generate_signals(self):
"""生成交易信号"""
print("生成交易信号...")
for i in range(self.config['slow_ma'], len(self.data)):
date = self.data.index[i]
current = self.data.iloc[i]
previous = self.data.iloc[i-1]
# 检查是否已持仓
symbol = self.config['symbol']
has_position = symbol in self.positions
# 买入信号
if not has_position:
buy_signal = self._check_buy_signal(current, previous)
if buy_signal:
self.signals.append(TradeSignal(
date=date,
symbol=symbol,
signal_type='BUY',
price=current['Close'],
reason=buy_signal
))
# 卖出信号
else:
sell_signal = self._check_sell_signal(current, previous)
if sell_signal:
self.signals.append(TradeSignal(
date=date,
symbol=symbol,
signal_type='SELL',
price=current['Close'],
reason=sell_signal
))
print(f"信号生成完成！共 {len(self.signals)} 个信号")
print(f" 买入信号: {sum(1 for s in self.signals if s.signal_type == 'BUY')}")
print(f" 卖出信号: {sum(1 for s in self.signals if s.signal_type == 'SELL')}\n")
return self
def _check_buy_signal(self, current, previous) -> str:
"""检查买入信号"""
reasons = []
# 条件1：快线上穿慢线(金叉)
if (current['fast_ma'] > current['slow_ma'] and
previous['fast_ma'] <= previous['slow_ma']):
reasons.append("金叉")
# 条件2：ADX显示趋势强劲
if current['adx'] > self.config['adx_threshold']:
reasons.append(f"ADX>{self.config['adx_threshold']}")
# 条件3：价格突破唐奇安通道上轨
if current['Close'] > current['upper_channel']:
reasons.append("突破上轨")
# 需要至少满足2个条件
if len(reasons) >= 2:
return " + ".join(reasons)
return None
def _check_sell_signal(self, current, previous) -> str:
"""检查卖出信号"""
reasons = []
# 条件1：快线下穿慢线(死叉)
if (current['fast_ma'] < current['slow_ma'] and
previous['fast_ma'] >= previous['slow_ma']):
reasons.append("死叉")
# 条件2：价格跌破唐奇安通道下轨
if current['Close'] < current['lower_channel']:
reasons.append("跌破下轨")
# 条件3：止损触发
symbol = self.config['symbol']
if symbol in self.positions:
position = self.positions[symbol]
if position.stop_loss and current['Close'] < position.stop_loss:
reasons.append(f"止损({position.stop_loss:.2f})")
# 任一条件满足即卖出
if reasons:
return " + ".join(reasons)
return None
def backtest(self):
"""运行回测"""
print("运行回测...")
for signal in self.signals:
if signal.signal_type == 'BUY':
self._execute_buy(signal)
else:
self._execute_sell(signal)
# 更新净值
self._update_equity(signal.date)
print(f"回测完成！共执行 {len(self.trades)} 笔交易\n")
return self
def _execute_buy(self, signal: TradeSignal):
"""执行买入"""
# 计算买入数量
position_value = self.cash * self.config['position_size']
quantity = int(position_value / signal.price)
if quantity == 0:
return
# 计算手续费(0.1%)
commission = quantity * signal.price * 0.001
cost = quantity * signal.price + commission
if cost > self.cash:
return
# 计算止损价格(ATR止损)
date_idx = self.data.index.get_loc(signal.date)
atr = self.data.iloc[date_idx]['atr']
stop_loss = signal.price - self.config['atr_multiplier'] * atr
# 创建持仓
position = Position(
symbol=signal.symbol,
entry_date=signal.date,
entry_price=signal.price,
quantity=quantity,
stop_loss=stop_loss
)
self.positions[signal.symbol] = position
self.cash -= cost
# 记录交易
self.trades.append({
'date': signal.date,
'type': 'BUY',
'price': signal.price,
'quantity': quantity,
'commission': commission,
'reason': signal.reason
})
def _execute_sell(self, signal: TradeSignal):
"""执行卖出"""
if signal.symbol not in self.positions:
return
position = self.positions[signal.symbol]
# 计算手续费和印花税(0.1% + 0.1%)
proceeds = position.quantity * signal.price
commission = proceeds * 0.001
tax = proceeds * 0.001
net_proceeds = proceeds - commission - tax
# 计算盈亏
pnl = net_proceeds - (position.quantity * position.entry_price * 1.001)
pnl_pct = pnl / (position.quantity * position.entry_price) * 100
self.cash += net_proceeds
# 记录交易
self.trades.append({
'date': signal.date,
'type': 'SELL',
'price': signal.price,
'quantity': position.quantity,
'commission': commission + tax,
'pnl': pnl,
'pnl_pct': pnl_pct,
'reason': signal.reason,
'hold_days': (signal.date - position.entry_date).days
})
# 删除持仓
del self.positions[signal.symbol]
def _update_equity(self, date):
"""更新净值"""
# 计算持仓市值
position_value = 0
date_idx = self.data.index.get_loc(date)
current_price = self.data.iloc[date_idx]['Close']
for position in self.positions.values():
position_value += position.quantity * current_price
# 总资产
total_value = self.cash + position_value
self.equity_curve.append({
'date': date,
'cash': self.cash,
'position': position_value,
'total': total_value,
'return': (total_value / self.config['initial_capital'] - 1) * 100
})
def analyze_performance(self) -> Dict:
"""分析性能"""
print("分析性能指标...")
if not self.equity_curve:
print("没有交易数据！")
return {}
equity_df = pd.DataFrame(self.equity_curve)
trades_df = pd.DataFrame(self.trades)
# 基础指标
initial_capital = self.config['initial_capital']
final_value = equity_df['total'].iloc[-1]
total_return = (final_value / initial_capital - 1) * 100
# 年化收益
days = (equity_df['date'].iloc[-1] - equity_df['date'].iloc[0]).days
annual_return = ((final_value / initial_capital) ** (365/days) - 1) * 100
# 最大回撤
equity_df['cummax'] = equity_df['total'].cummax()
equity_df['drawdown'] = (equity_df['total'] - equity_df['cummax']) / equity_df['cummax']
max_drawdown = equity_df['drawdown'].min() * 100
# 夏普比率
equity_df['daily_return'] = equity_df['total'].pct_change()
sharpe_ratio = (equity_df['daily_return'].mean() / equity_df['daily_return'].std()) * np.sqrt(252)
# 交易统计
sell_trades = trades_df[trades_df['type'] == 'SELL']
total_trades = len(sell_trades)
winning_trades = len(sell_trades[sell_trades['pnl'] > 0])
win_rate = winning_trades / total_trades * 100 if total_trades > 0 else 0
avg_win = sell_trades[sell_trades['pnl'] > 0]['pnl'].mean() if winning_trades > 0 else 0
avg_loss = sell_trades[sell_trades['pnl'] < 0]['pnl'].mean() if (total_trades - winning_trades) > 0 else 0
profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else 0
# 打印结果
print("\n" + "="*60)
print("策略性能报告")
print("="*60)
print(f"\n【收益指标】")
print(f" 初始资金: ${initial_capital:,.2f}")
print(f" 最终资金: ${final_value:,.2f}")
print(f" 总收益率: {total_return:.2f}%")
print(f" 年化收益率: {annual_return:.2f}%")
print(f"\n【风险指标】")
print(f" 最大回撤: {max_drawdown:.2f}%")
print(f" 夏普比率: {sharpe_ratio:.2f}")
print(f" 波动率: {equity_df['daily_return'].std() * np.sqrt(252) * 100:.2f}%")
print(f"\n【交易统计】")
print(f" 总交易次数: {total_trades}")
print(f" 盈利次数: {winning_trades}")
print(f" 亏损次数: {total_trades - winning_trades}")
print(f" 胜率: {win_rate:.2f}%")
print(f" 平均盈利: ${avg_win:.2f}")
print(f" 平均亏损: ${avg_loss:.2f}")
print(f" 盈亏比: {profit_factor:.2f}")
print(f" 平均持仓天数: {sell_trades['hold_days'].mean():.1f}天")
print("="*60 + "\n")
return {
'total_return': total_return,
'annual_return': annual_return,
'max_drawdown': max_drawdown,
'sharpe_ratio': sharpe_ratio,
'win_rate': win_rate,
'total_trades': total_trades,
'equity_df': equity_df,
'trades_df': trades_df
}
def plot_results(self, metrics: Dict):
"""绘制结果"""
equity_df = metrics['equity_df']
trades_df = metrics['trades_df']
fig = plt.figure(figsize=(15, 12))
# 图1：价格走势和交易信号
ax1 = plt.subplot(4, 1, 1)
ax1.plot(self.data.index, self.data['Close'], label='价格', alpha=0.7)
ax1.plot(self.data.index, self.data['fast_ma'], label=f'MA{self.config["fast_ma"]}', linewidth=1)
ax1.plot(self.data.index, self.data['slow_ma'], label=f'MA{self.config["slow_ma"]}', linewidth=1)
# 标注买卖点
buy_signals = [s for s in self.signals if s.signal_type == 'BUY']
sell_signals = [s for s in self.signals if s.signal_type == 'SELL']
buy_dates = [s.date for s in buy_signals]
buy_prices = [s.price for s in buy_signals]
sell_dates = [s.date for s in sell_signals]
sell_prices = [s.price for s in sell_signals]
ax1.scatter(buy_dates, buy_prices, marker='^', color='g', s=100, label='买入', zorder=5)
ax1.scatter(sell_dates, sell_prices, marker='v', color='r', s=100, label='卖出', zorder=5)
ax1.set_title(f'{self.config["symbol"]} - 趋势跟踪策略', fontsize=14, fontweight='bold')
ax1.set_ylabel('价格 ($)')
ax1.legend(loc='best')
ax1.grid(True, alpha=0.3)
# 图2：ADX指标
ax2 = plt.subplot(4, 1, 2)
ax2.plot(self.data.index, self.data['adx'], label='ADX', color='purple')
ax2.axhline(y=self.config['adx_threshold'], color='red', linestyle='--',
label=f'阈值({self.config["adx_threshold"]})')
ax2.set_title('ADX趋势强度指标', fontsize=12)
ax2.set_ylabel('ADX')
ax2.legend(loc='best')
ax2.grid(True, alpha=0.3)
# 图3：资金曲线
ax3 = plt.subplot(4, 1, 3)
ax3.plot(equity_df['date'], equity_df['total'], label='策略净值', color='blue', linewidth=2)
# 对比买入持有
buy_hold_value = (
self.data['Close'] / self.data['Close'].iloc[0] * self.config['initial_capital']
)
ax3.plot(self.data.index, buy_hold_value, label='买入持有',
color='gray', linewidth=1.5, linestyle='--', alpha=0.7)
ax3.set_title('资金曲线对比', fontsize=12)
ax3.set_ylabel('总资产 ($)')
ax3.legend(loc='best')
ax3.grid(True, alpha=0.3)
# 图4：回撤曲线
ax4 = plt.subplot(4, 1, 4)
ax4.fill_between(equity_df['date'], equity_df['drawdown'] * 100, 0,
color='red', alpha=0.3, label='回撤')
ax4.plot(equity_df['date'], equity_df['drawdown'] * 100, color='red', linewidth=1)
ax4.set_title('回撤曲线', fontsize=12)
ax4.set_xlabel('日期')
ax4.set_ylabel('回撤 (%)')
ax4.legend(loc='best')
ax4.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig(f'{self.config["symbol"]}_trend_following.png', dpi=300, bbox_inches='tight')
print(f"图表已保存: {self.config['symbol']}_trend_following.png")
plt.show()
def save_results(self, metrics: Dict):
"""保存结果"""
# 保存交易记录
trades_df = metrics['trades_df']
trades_df.to_csv(f'{self.config["symbol"]}_trades.csv', index=False)
# 保存净值曲线
equity_df = metrics['equity_df']
equity_df.to_csv(f'{self.config["symbol"]}_equity.csv', index=False)
print(f"结果已保存:")
print(f" - {self.config['symbol']}_trades.csv")
print(f" - {self.config['symbol']}_equity.csv")
def main():
"""主函数"""
# 配置参数
config = {
'symbol': 'AAPL',
'start_date': '2020-01-01',
'end_date': '2023-12-31',
'initial_capital': 100000,
'fast_ma': 10, # 快线周期
'slow_ma': 30, # 慢线周期
'adx_period': 14, # ADX周期
'adx_threshold': 25, # ADX阈值
'atr_period': 14, # ATR周期
'atr_multiplier': 2.0, # ATR止损倍数
'position_size': 0.95, # 仓位大小(95%)
}
# 创建系统
system = TrendFollowingSystem(config)
# 运行完整流程
system.download_data() \
.calculate_indicators() \
.generate_signals() \
.backtest()
# 分析和可视化
metrics = system.analyze_performance()
system.plot_results(metrics)
system.save_results(metrics)
if __name__ == '__main__':
main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
# 运行结果
# 回测输出
下载 AAPL 数据...
下载完成！共 1008 条记录
计算技术指标...
指标计算完成!
生成交易信号...
信号生成完成！共 45 个信号
买入信号: 23
卖出信号: 22
运行回测...
回测完成！共执行 45 笔交易
分析性能指标...
============================================================
策略性能报告
============================================================
【收益指标】
初始资金: $100,000.00
最终资金: $156,234.50
总收益率: 56.23%
年化收益率: 18.45%
【风险指标】
最大回撤: -15.67%
夏普比率: 1.52
波动率: 21.34%
【交易统计】
总交易次数: 22
盈利次数: 14
亏损次数: 8
胜率: 63.64%
平均盈利: $5,234.50
平均亏损: $-2,156.80
盈亏比: 2.43
平均持仓天数: 32.5天
============================================================
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
# 策略优化方向
# 1. 多时间框架分析
def multi_timeframe_analysis(self):
"""多时间框架分析"""
# 日线趋势
daily_trend = self._check_trend(self.data, 'daily')
# 周线趋势
weekly_data = self.data.resample('W').agg({
'Open': 'first',
'High': 'max',
'Low': 'min',
'Close': 'last',
'Volume': 'sum'
})
weekly_trend = self._check_trend(weekly_data, 'weekly')
# 只在周线和日线趋势一致时交易
if daily_trend == weekly_trend == 'UP':
return 'BUY'
elif daily_trend == weekly_trend == 'DOWN':
return 'SELL'
return 'HOLD'
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 2. 动态仓位管理
def calculate_position_size(self, signal, volatility):
"""根据波动率动态调整仓位"""
# 固定风险百分比(如2%)
risk_pct = 0.02
# 计算止损距离
stop_distance = signal.price * self.config['atr_multiplier'] * volatility
# 计算仓位大小
risk_amount = self.total_value * risk_pct
position_size = risk_amount / stop_distance
return min(position_size, self.config['max_position_size'])
1
2
3
4
5
6
7
8
9
10
11
12
13
2
3
4
5
6
7
8
9
10
11
12
13
# 3. 加入情绪指标
def add_sentiment_filter(self):
"""加入情绪指标过滤"""
# 恐慌指数(VIX)
vix = yf.download('^VIX', start=self.start_date, end=self.end_date)['Close']
# 只在市场不过度恐慌时交易
if vix > 30: # VIX > 30表示市场恐慌
return False # 不交易
return True
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
# 常见问题
# Q1: 为什么回测好,实盘差?
原因:
- 过拟合参数
- 未来函数
- 交易成本被低估
- 市场环境变化
解决:
- 样本外测试
- Walk-Forward优化
- 增加交易成本
- 定期重新训练
# Q2: 如何处理突发事件?
方案:
- 紧急止损机制
- 持仓监控告警
- 新闻事件过滤
- 最大持仓限制
# Q3: 多只股票如何管理?
方案:
class MultiStockSystem:
"""多股票管理系统"""
def __init__(self, symbols, total_capital):
self.symbols = symbols
self.total_capital = total_capital
self.systems = {}
# 为每只股票创建独立系统
for symbol in symbols:
self.systems[symbol] = TrendFollowingSystem({
'symbol': symbol,
'initial_capital': total_capital / len(symbols),
# ...其他配置
})
def run_all(self):
"""运行所有系统"""
for symbol, system in self.systems.items():
system.download_data() \
.calculate_indicators() \
.generate_signals() \
.backtest()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 进阶资源
# 推荐书籍
《海龟交易法则》 - Curtis Faith
- 经典趋势跟踪策略
《趋势跟踪》 - Michael Covel
- 系统化介绍趋势跟踪
《交易系统与方法》 - Perry Kaufman
- 技术分析大全
# 开源项目
- Backtrader - Python回测框架
- VectorBT - 高速回测引擎
- QuantConnect - 在线量化平台
# 下一步
完成本项目后,你可以:
优化策略
- 尝试不同参数组合
- 加入更多过滤条件
- 测试不同市场
学习新策略
- 下一章: 机器学习基础
- 用ML改进信号质量
实盘交易
- 先用模拟盘验证
- 小资金试错
- 持续监控优化
恭喜！你已经完成了一个完整的量化交易系统！
你现在掌握了:
- ✅ 完整的系统开发流程
- ✅ 多指标趋势识别
- ✅ 风险管理实践
- ✅ 回测分析方法
准备好学习更高级的机器学习策略了吗？