|
以下是一個基于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['ma_long']) | (df['macd'] < df['macd_signal']) | (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. **買入信號四重驗證**： - **均線金叉**：MA5上穿MA20，確認短期趨勢轉強 - **MACD共振**：快線上穿慢線且柱狀線轉正 - **RSI中性**：避免在超買區追高 - **量能配合**：成交量突破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.position.price * self.stop_loss: 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
``` 3. **流動性風險**：實際交易中需考慮盤口價差 4. **數據頻率**：日線策略需注意財報/事件沖擊 該程序可通過Jupyter Notebook實時運行，建議先用歷史數據驗證策略有效性，再結合實時行情接口（如Tushare Pro）部署到生產環境。 |
|
|
