摘要：以下是一个基于Python的股票量化交易程序，结合均线系统、macd金叉、rsi指标和成交量，实现买卖信号的精准捕捉。代码包含数据获取、指标计算、信号生成及可视化全流程：
以下是一个基于Python的股票量化交易程序，结合均线系统、macd金叉、rsi指标和成交量，实现买卖信号的精准捕捉。代码包含数据获取、指标计算、信号生成及可视化全流程：
```python
import pandas as pd
import numpy as np
import akshare as ak
import matplotlib.pyplot as plt
from talib import abstract
from backtrader import Cerebro, feeds
# 参数配置
PARAMS = {
"ma_short": 5, # 短期均线
"ma_long": 20, # 长期均线
"rsi_period": 14, # RSI周期
"volume_ma": 20, # 成交量均线周期
"macd_fast": 12, # MACD快线
"macd_slow": 26, # MACD慢线
"macd_signal": 9 # MACD信号线
}
# 数据获取与预处理
def get_stock_data(code, start_date="20220101"):
"""获取复权历史数据"""
df = ak.stock_zh_a_hist(symbol=code, adjust="hfq")
df['date'] = pd.to_datetime(df['日期'])
df = df[['date', '开盘', '最高', '最低', '收盘', '成交量']]
df.columns = ['date', 'open', 'high', 'low', 'close', 'volume']
df.set_index('date', inplace=True)
return df[start_date:]
# 技术指标计算
def calculate_indicators(df):
"""计算关键指标"""
# 均线系统
df['ma_short'] = df['close'].rolling(PARAMS['ma_short']).mean
df['ma_long'] = df['close'].rolling(PARAMS['ma_long']).mean
# MACD指标
macd = abstract.MACD(df,
fastperiod=PARAMS['macd_fast'],
slowperiod=PARAMS['macd_slow'],
signalperiod=PARAMS['macd_signal'])
df['macd'] = macd['macd']
df['macd_signal'] = macd['macd_signal']
df['macd_hist'] = macd['macd_hist']
# RSI指标
df['rsi'] = abstract.RSI(df, timeperiod=PARAMS['rsi_period'])
# 成交量均线
df['vol_ma'] = df['volume'].rolling(PARAMS['volume_ma']).mean
return df.dropna
# 交易信号生成
def generate_signals(df):
"""生成买卖信号"""
df['signal'] = 0
# 买入条件（同时满足）：
# 1. 短期均线上穿长期均线
# 2. MACD金叉（快线上穿慢线）
# 3. RSI在30-70之间
# 4. 成交量突破均量线
buy_cond = (
(df['ma_short'] > df['ma_long']) &
(df['macd'] > df['macd_signal']) &
(df['rsi'].between(30, 70)) &
(df['volume'] > df['vol_ma'])
)
# 卖出条件（满足其一）：
# 1. 短期均线下穿长期均线
# 2. MACD死叉（快线下穿慢线）
# 3. RSI超买（>70）
sell_cond = (
(df['ma_short']
(df['macd']
(df['rsi'] > 70)
)
df.loc[buy_cond, 'signal'] = 1 # 买入信号
df.loc[sell_cond, 'signal'] = -1 # 卖出信号
return df
# 可视化分析
def visualize_signals(df, code):
plt.figure(figsize=(16,12))
# 价格与均线
ax1 = plt.subplot(411)
df[['close','ma_short','ma_long']].plot(ax=ax1)
ax1.scatter(df.index[df['signal']==1], df['close'][df['signal']==1],
marker='^', color='g', label='Buy')
ax1.scatter(df.index[df['signal']==-1], df['close'][df['signal']==-1],
marker='v', color='r', label='Sell')
ax1.set_title(f"{code} Price with Signals")
# MACD
ax2 = plt.subplot(412)
df[['macd','macd_signal']].plot(ax=ax2)
ax2.bar(df.index, df['macd_hist'], color=np.where(df['macd_hist']>0, 'g','r'))
ax2.set_title("MACD")
# RSI
ax3 = plt.subplot(413)
df['rsi'].plot(ax=ax3)
ax3.axhline(70, ls='--', color='r')
ax3.axhline(30, ls='--', color='g')
ax3.set_title("RSI")
# 成交量
ax4 = plt.subplot(414)
df[['volume','vol_ma']].plot(ax=ax4)
ax4.set_title("Volume")
plt.tight_layout
plt.show
# 回测策略
class MultiIndicatorStrategy(bt.Strategy):
params = (
('risk_per_trade', 0.02), # 单笔风险2%
)
def __init__(self):
self.order = None
self.signal = self.datas[0].signal
def next(self):
if self.order: # 检查是否有挂单
return
# 开仓逻辑
if self.signal[0] == 1 and not self.position:
size = self.broker.getvalue * self.params.risk_per_trade / self.data.close[0]
self.buy(size=size)
# 平仓逻辑
elif self.signal[0] == -1 and self.position:
self.close
# 主程序
if __name__ == "__main__":
# 示例股票代码（贵州茅台）
stock_code = "600519"
df = get_stock_data(stock_code)
df = calculate_indicators(df)
df = generate_signals(df)
# 可视化信号
visualize_signals(df, stock_code)
# 回测引擎
cerebro = bt.Cerebro
data = feeds.PandasData(dataname=df)
cerebro.adddata(data)
cerebro.addstrategy(MultiIndicatorStrategy)
# 设置初始资金
cerebro.broker.setcash(100000.0)
# 执行回测
print('初始资金: %.2f' % cerebro.broker.getvalue)
cerebro.run
print('最终资金: %.2f' % cerebro.broker.getvalue)
```
---
### **策略逻辑说明**
1. **买入信号四重验证**：
- **量能配合**：成交量突破20日均量，确认资金入场
2. **卖出信号三重保险**：
- 均线死叉（趋势反转）
- MACD死叉（动量衰减）
- RSI超买（短期过热）
3. **风险管理**：
- 单笔交易风险控制在总资金2%
- 强制止损（可在策略中添加）
---
### **优化方向**
1. **动态参数优化**：
```python
# 使用网格搜索寻找最优参数组合
param_grid = {
'ma_short': [5, 10, 20],
'ma_long': [20, 30, 60],
'rsi_period': [14, 21, 28]
}
```
2. **加入止损机制**：
```python
# 在策略类中添加
def __init__(self):
self.stop_loss = 0.95 # 5%止损
def next(self):
if self.position:
if self.data.close[0]
self.close
```
3. **多周期验证**：
```python
# 同时分析周线级别趋势
weekly_df = df.resample('W').last
weekly_df = calculate_indicators(weekly_df)
```
4. **市场状态过滤**：
```python
# 增加大盘趋势判断
index_data = get_stock_data('000001') # 上证指数
market_up = index_data['close'] > index_data['ma_20']
```
---
### **注意事项**
1. **避免过度拟合**：参数优化需在样本外数据验证
2. **交易成本**：回测中需计入手续费（默认0.1%）
```python
cerebro.broker.setcommission(commission=0.001) # 设置佣金
```
3. **流动性风险**：实际交易中需考虑盘口价差
4. **数据频率**：日线策略需注意财报/事件冲击
该程序可通过Jupyter Notebook实时运行，建议先用历史数据验证策略有效性，再结合实时行情接口（如Tushare Pro）部署到生产环境。
来源：向阳随笔