項目)
短線選股絕招保姆級教程:從零搭建量化實戰(zhàn)項目
看了一堆教程還是不會寫項目?別急,這篇短線選股絕招保姆級教程帶你從零搭建。
項目目標(biāo)與痛點直擊
很多開發(fā)者朋友在GitHub上收藏了幾百個“量化交易”項目,代碼看著都懂,真動手跑起來就報錯,或者邏輯完全無法落地。核心痛點在于:教程只講語法,不講工程化;只給結(jié)果,不給推導(dǎo)過程。
為了打破這個僵局,我們基于Python構(gòu)建一個可復(fù)現(xiàn)、可回測的短線選股系統(tǒng)。目標(biāo)不是讓你立刻賺大錢,而是讓你掌握數(shù)據(jù)獲取、因子計算、策略回測、風(fēng)險控制的全鏈路工程能力。
本項目參考了vnpy官方源碼倉庫中關(guān)于回測引擎的設(shè)計思想,但做了極致的簡化,確保初學(xué)者能在半小時內(nèi)跑通核心邏輯。
目錄結(jié)構(gòu)設(shè)計
一個規(guī)范的工程項目,結(jié)構(gòu)清晰是第一步。我們采用模塊化設(shè)計,避免所有代碼堆在一個文件里。
stock-picker/
├── config/
│ └── settings.py # 配置文件:API密鑰、回測參數(shù)
├── data/
│ ├── fetcher.py # 數(shù)據(jù)獲取模塊
│ └── processor.py # 數(shù)據(jù)清洗與預(yù)處理
├── strategy/
│ └── short_term.py # 核心策略:短線選股邏輯
├── backtest/
│ └── engine.py # 簡易回測引擎
├── utils/
│ └── logger.py # 日志工具
├── main.py # 主入口
└── requirements.txt # 依賴庫這種結(jié)構(gòu)的好處是:數(shù)據(jù)、策略、回測解耦。當(dāng)你想更換數(shù)據(jù)源時,只需修改data/fetcher.py,策略代碼無需變動。這就是工程化思維的體現(xiàn)。
核心代碼實現(xiàn):短線因子計算
短線選股的核心在于動量與波動率的平衡。我們選取兩個經(jīng)典因子:5日收益率(Momentum) 和 20日波動率(Volatility)。
1. 數(shù)據(jù)獲取模塊
這里我們使用akshare庫,它是國內(nèi)開發(fā)者獲取A股數(shù)據(jù)最友好的開源庫之一。
# data/fetcher.py
import akshare as ak
import pandas as pd
from datetime import datetime, timedeltaclass DataFetcher:def __init__(self, symbol: str):初始化數(shù)據(jù)獲取器:param symbol: 股票代碼,如 '600519'self.symbol = symbolself.df = Nonedef fetch_daily_data(self, days: int = 100) - pd.DataFrame:獲取最近N天的日線數(shù)據(jù):param days: 獲取天數(shù):return: 包含OHLCV的DataFrameend_date = datetime.now().strftime('%Y%m%d')start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')try:# 獲取前復(fù)權(quán)數(shù)據(jù),避免除權(quán)除息導(dǎo)致的股價斷崖self.df = ak.stock_zh_a_hist(symbol=self.symbol, period=daily, start_date=start_date, end_date=end_date,adjust=qfq)# 標(biāo)準(zhǔn)化列名,統(tǒng)一為小寫英文self.df.columns = ['date', 'code', 'open', 'high', 'low', 'close', 'volume', 'amount', 'amplitude', 'pct_chg', 'vol_ratio']self.df['date'] = pd.to_datetime(self.df['date'])self.df.set_index('date', inplace=True)return self.dfexcept Exception as e:print(f數(shù)據(jù)獲取失敗: {e})return pd.DataFrame()關(guān)鍵點解析:前復(fù)權(quán)(qfq):短線交易必須使用復(fù)權(quán)數(shù)據(jù),否則歷史K線在分紅送股后會出現(xiàn)虛假的下跌信號,導(dǎo)致策略失效。
異常處理:網(wǎng)絡(luò)請求不穩(wěn)定是常態(tài),必須捕獲異常并返回空DataFrame,防止程序崩潰。2. 因子計算與策略邏輯
這是項目的靈魂。我們定義一個簡單的打分系統(tǒng):動量越強得分越高,波動率適中(不過高)得分越高。
# strategy/short_term.py
import pandas as pd
import numpy as npclass ShortTermStrategy:def __init__(self, momentum_window: int = 5, vol_window: int = 20)::param momentum_window: 動量計算窗口(天):param vol_window: 波動率計算窗口(天)self.momentum_window = momentum_windowself.vol_window = vol_windowdef calculate_factors(self, df: pd.DataFrame) - pd.DataFrame:計算核心因子并打分# 1. 計算日收益率df['daily_return'] = df['close'].pct_change()# 2. 計算動量因子:最近5天的累計收益率df['momentum'] = df['close'].pct_change(self.momentum_window)# 3. 計算波動率因子:最近20天收益率的標(biāo)準(zhǔn)差df['volatility'] = df['daily_return'].rolling(window=self.vol_window).std()# 4. 數(shù)據(jù)清洗:去除NaN值df.dropna(subset=['momentum', 'volatility'], inplace=True)# 5. 標(biāo)準(zhǔn)化處理(Z-Score),使不同量綱的因子可比# 動量越大越好,波動率越低越好(短線規(guī)避高波動風(fēng)險)df['momentum_score'] = (df['momentum'] - df['momentum'].mean()) / df['momentum'].std()df['volatility_score'] = (df['volatility'].mean() - df['volatility']) / df['volatility'].std()# 6. 綜合得分:動量權(quán)重60%,波動率權(quán)重40%df['composite_score'] = 0.6 * df['momentum_score'] + 0.4 * df['volatility_score']return dfdef select_stock(self, df: pd.DataFrame, top_n: int = 1) - list:根據(jù)綜合得分選出前N只股票# 取最后一條有效數(shù)據(jù)(即最新交易日)latest_data = df.iloc[-1]# 實際應(yīng)用中應(yīng)在全市場范圍內(nèi)排序,此處簡化為單股示例# 若得分大于0,認(rèn)為具備短線買入價值if latest_data['composite_score'] 0:return [latest_data['code']]return []避坑指南:Z-Score標(biāo)準(zhǔn)化:直接相加不同因子的原始值是沒有意義的,因為收益率是百分比,波動率也是百分比但量級不同。標(biāo)準(zhǔn)化后,分?jǐn)?shù)代表“偏離平均水平的程度”,這才是可比的。
Rolling窗口:rolling(window=20).std() 是滾動標(biāo)準(zhǔn)差,它反映了最近20天的波動情況,而不是全周期的波動,這對短線至關(guān)重要。運行與測試:簡易回測引擎
有了策略,必須驗證其有效性。我們寫一個極簡的回測引擎,模擬從2023年1月1日至今的交易。
# backtest/engine.py
import pandas as pdclass SimpleBacktester:def __init__(self, initial_capital: float = 100000):self.initial_capital = initial_capitalself.current_capital = initial_capitalself.trades = []def run(self, strategy, data_fetcher):執(zhí)行回測:param strategy: 策略實例:param data_fetcher: 數(shù)據(jù)獲取實例df = data_fetcher.fetch_daily_data(days=365)if df.empty:return# 預(yù)計算所有歷史因子df = strategy.calculate_factors(df)# 遍歷每一個交易日(跳過前20天,因為波動率需要20天數(shù)據(jù))for i in range(20, len(df)):current_date = df.index[i]current_price = df.iloc[i]['close']# 獲取截至當(dāng)前日期的歷史數(shù)據(jù)(防止未來函數(shù)?。﹉ist_df = df.iloc[:i+1]# 獲取信號signal = strategy.select_stock(hist_df, top_n=1)if signal:# 簡化邏輯:次日開盤價買入if i + 1 len(df):buy_price = df.iloc[i+1]['open']# 假設(shè)全倉買入shares = self.current_capital / buy_price# 扣除手續(xù)費(萬3)fee = shares * buy_price * 0.0003self.current_capital -= (shares * buy_price + fee)# 假設(shè)持有5天后賣出sell_day = i + 6if sell_day len(df):sell_price = df.iloc[sell_day]['close']sell_fee = shares * sell_price * 0.0003self.current_capital += (shares * sell_price - sell_fee)# 記錄交易self.trades.append({'buy_date': df.index[i+1],'sell_date': df.index[sell_day],'buy_price': buy_price,'sell_price': sell_price,'pnl': (sell_price - buy_price) * shares - fee - sell_fee})# 輸出結(jié)果self.report()def report(self):生成回測報告if not self.trades:print(無交易記錄)returntotal_pnl = sum(t['pnl'] for t in self.trades)win_rate = len([t for t in self.trades if t['pnl'] 0]) / len(self.trades) * 100max_drawdown = self.calculate_max_drawdown()print(f初始資金: {self.initial_capital:,.2f})print(f期末資金: {self.current_capital:,.2f})print(f總收益: {total_pnl:,.2f})print(f勝率: {win_rate:.2f}%)print(f最大回撤: {max_drawdown:.2f}%)def calculate_max_drawdown(self):# 此處簡化,實際應(yīng)基于凈值曲線計算return 10.0 核心概念:未來函數(shù)
在回測中,hist_df = df.iloc[:i+1] 這一行至關(guān)重要。它確保在第i天做決策時,只能看到第i天及之前的數(shù)據(jù)。如果你用了第i+1天的收盤價來計算第i天的因子,這就是“未來函數(shù)”,會導(dǎo)致回測結(jié)果虛高,實盤必虧。
優(yōu)化擴展與避坑指南
跑通基礎(chǔ)版后,你會遇到幾個實際問題,這也是從“教程”到“工程”的跨越點。
1. 數(shù)據(jù)缺失處理
A股每天都有停牌股。如果在停牌日,close價格為NaN,或者成交量為0。解決方案:在processor.py中增加過濾邏輯。def clean_data(df):# 過濾成交量為0的停牌日df = df[df['volume'] 0]# 前向填充少量缺失值df['close'].fillna(method='ffill', inplace=True)return df2. 參數(shù)過擬合
你可能發(fā)現(xiàn),把momentum_window從5改成3,收益率翻倍。這很可能是過擬合。解決方案:進(jìn)行參數(shù)敏感性分析。不要只測試一個參數(shù),而是測試一個范圍(如3-10天),觀察策略績效是否穩(wěn)定。如果只在某個特定參數(shù)下表現(xiàn)好,說明策略不可靠。3. 實盤對接
回測是歷史,實盤是現(xiàn)實。延遲問題:行情數(shù)據(jù)通常有毫秒級延遲,你的策略信號發(fā)出后,價格可能已經(jīng)變了。
滑點:回測中假設(shè)按收盤價成交,實盤中可能因流動性不足無法成交,或成交價偏離。建議在回測中加入slippage參數(shù),例如buy_price = open * 1.001。4. 日志與監(jiān)控
在生產(chǎn)環(huán)境中,必須記錄每一步?jīng)Q策。
# utils/logger.py
import logging
import osdef setup_logger(name=stock_picker):if not os.path.exists(logs):os.makedirs(logs)logging.basicConfig(filename=flogs/{name}.log,level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')return logging.getLogger(name)在策略每次觸發(fā)買賣時,調(diào)用logger.info(fBuy signal at {price})。這是排查實盤bug的唯一線索。
小結(jié)
這篇短線選股絕招保姆級教程,帶你完成了從數(shù)據(jù)獲取到因子構(gòu)建,再到回測驗證的完整閉環(huán)。
回顧一下我們學(xué)到的核心工程能力:模塊化設(shè)計:數(shù)據(jù)、策略、回測分離,便于維護(hù)。
因子標(biāo)準(zhǔn)化:Z-Score讓不同維度的指標(biāo)可比。
防止未來函數(shù):回測中嚴(yán)格限制數(shù)據(jù)可見范圍。
異常處理與日志:讓程序在臟數(shù)據(jù)和網(wǎng)絡(luò)波動下依然健壯。這個項目的代碼框架已經(jīng)搭建完畢,你可以基于此替換因子(如加入RSI、MACD),或更換數(shù)據(jù)源(如從akshare切換到Tushare)。真正的量化交易,始于代碼,終于風(fēng)控。
這個知識點你面試被問過嗎?比如“如何防止回測中的未來函數(shù)”或“因子標(biāo)準(zhǔn)化為什么要用Z-Score而不是Min-Max”,留言說說你的看法或遇到的坑,我們一起交流。