簡介這是一套基于Python構(gòu)建的全棧股票分析系統(tǒng)源碼面向金融數(shù)據(jù)分析學(xué)習(xí)者、量化研究入門者及需要搭建個(gè)性化行情工具的開發(fā)者。系統(tǒng)以akshare為核心數(shù)據(jù)接口覆蓋A股、港股、美股等多市場行情并整合pandas、TensorFlow、tornado等工具實(shí)現(xiàn)數(shù)據(jù)采集、統(tǒng)計(jì)建模與可視化展示的完整鏈路同時(shí)支持實(shí)時(shí)追蹤與歷史回溯。資源包共172個(gè)文件約2.18MB包含29個(gè)py核心腳本、34個(gè)js與19個(gè)html/css前端頁面、28個(gè)zbak備份文件以及dockerfile、conf、sh等部署配置目錄結(jié)構(gòu)清晰便于按模塊閱讀與二次開發(fā)。已有47人學(xué)習(xí)下載。代碼經(jīng)過重構(gòu)模塊化程度較高pandas強(qiáng)化了時(shí)間序列清洗能力tornado異步架構(gòu)支撐高并發(fā)實(shí)時(shí)更新TensorFlow用于預(yù)測與風(fēng)險(xiǎn)評(píng)估建模適合作為金融分析工具鏈的實(shí)踐參考。1. 一套能跑起來的 Python 全棧股票分析系統(tǒng)到底由哪些模塊拼成很多人搜「基于Python的全棧股票分析系統(tǒng)源碼」腦子里想的是下載一個(gè)壓縮包、解壓、python app.py然后瀏覽器里就出現(xiàn)一個(gè)能看 K 線、能選股、能回測的完整系統(tǒng)?，F(xiàn)實(shí)是這類項(xiàng)目從來不是「一個(gè)文件跑天下」而是數(shù)據(jù)采集、數(shù)據(jù)存儲(chǔ)、指標(biāo)計(jì)算、策略回測、Web 接口、前端展示這幾塊拼起來的。全棧兩個(gè)字重點(diǎn)不在「全」而在「?！埂恳粚佣家塥?dú)立跑通再串成一條鏈路。這套系統(tǒng)適合兩類人一類是剛學(xué)完 Python 基礎(chǔ)語法、想找一個(gè)真實(shí)項(xiàng)目練手的開發(fā)者另一類是手里有選股想法、但缺一套可復(fù)用分析框架的從業(yè)者。它解決的核心問題是把「拉數(shù)據(jù) → 算指標(biāo) → 出信號(hào) → 看結(jié)果」這條鏈路固化下來而不是每次都在 Jupyter 里重新寫一遍。下面按模塊拆開講每一步都落到能復(fù)現(xiàn)的命令和代碼上。2. 數(shù)據(jù)層怎么搭從行情接口到本地緩存的完整鏈路2.1 為什么先定數(shù)據(jù)源再寫任何分析代碼股票分析系統(tǒng)的第一道坎不是算法是數(shù)據(jù)。常見做法是先用公開行情接口拉日線數(shù)據(jù)落到本地做緩存后續(xù)所有指標(biāo)計(jì)算和回測都讀本地文件。這樣做的好處是接口掛了不影響你調(diào)試策略回測速度也快一個(gè)量級(jí)。我一般會(huì)選akshare或tushare這類庫做數(shù)據(jù)入口前者免費(fèi)、覆蓋 A 股較全后者需要 token 但字段規(guī)范。選型時(shí)看三個(gè)點(diǎn)一是復(fù)權(quán)處理是否內(nèi)置二是歷史數(shù)據(jù)能拉多長三是請(qǐng)求頻率限制。日線級(jí)別分析前兩點(diǎn)比第三點(diǎn)重要因?yàn)槿站€數(shù)據(jù)一天只需要更新一次頻率壓力不大。把數(shù)據(jù)源定下來之后目錄結(jié)構(gòu)就要提前規(guī)劃否則后面指標(biāo)、回測、接口三層會(huì)互相污染。# 項(xiàng)目根目錄結(jié)構(gòu)先建好再寫代碼 stock-analysis/ ├── data/ # 本地行情緩存按股票代碼分文件 ├── indicators/ # 指標(biāo)計(jì)算模塊 ├── backtest/ # 回測引擎 ├── api/ # Web 接口層 ├── web/ # 前端頁面 └── config.py # 數(shù)據(jù)源、路徑、參數(shù)集中配置這個(gè)結(jié)構(gòu)的關(guān)鍵是data/和config.py分離。數(shù)據(jù)文件不進(jìn)版本控制配置集中管理換數(shù)據(jù)源時(shí)只改config.py一個(gè)地方。2.2 用 akshare 拉日線并落地成 parquet 的最小腳本下面這段代碼做三件事拉一只股票的日線、做前復(fù)權(quán)、存成 parquet。parquet 比 CSV 快列式存儲(chǔ)對(duì)指標(biāo)計(jì)算友好文件體積也小。import akshare as ak import pandas as pd from pathlib import Path DATA_DIR Path(data) DATA_DIR.mkdir(exist_okTrue) def fetch_daily(symbol: str, start: str, end: str) - pd.DataFrame: # symbol 格式如 000001akshare 需要不帶交易所前綴 df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_dateend, adjustqfq # 前復(fù)權(quán)回測必須用復(fù)權(quán)價(jià) ) # 統(tǒng)一列名避免后續(xù)模塊依賴中文列名 df df.rename(columns{ 日期: date, 開盤: open, 收盤: close, 最高: high, 最低: low, 成交量: volume }) df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) return df def save_daily(symbol: str, df: pd.DataFrame): path DATA_DIR / f{symbol}.parquet df.to_parquet(path, indexFalse) print(fsaved {len(df)} rows - {path}) if __name__ __main__: df fetch_daily(000001, 20200101, 20241231) save_daily(000001, df)邏輯說明adjustqfq是回測正確性的前提用不復(fù)權(quán)價(jià)算出來的收益率是錯(cuò)的。列名統(tǒng)一成英文是為了后面指標(biāo)模塊不依賴中文列名換數(shù)據(jù)源時(shí)只改這一個(gè)函數(shù)。參數(shù)上start_date和end_date用字符串傳入akshare 內(nèi)部會(huì)解析不用自己轉(zhuǎn) datetime。2.3 增量更新與數(shù)據(jù)校驗(yàn)的兩個(gè)必做檢查全量拉一次之后日常只需要增量更新。增量邏輯是讀本地 parquet 的最后一條日期從那天往后拉。這里有兩個(gè)坑一是停牌日會(huì)導(dǎo)致日期不連續(xù)二是接口偶爾返回空 DataFrame。所以更新函數(shù)里必須做校驗(yàn)。def update_daily(symbol: str): path DATA_DIR / f{symbol}.parquet if not path.exists(): df fetch_daily(symbol, 20200101, 20241231) save_daily(symbol, df) return old pd.read_parquet(path) last_date old[date].max().strftime(%Y%m%d) new fetch_daily(symbol, last_date, 20241231) if new.empty: print(f{symbol} no new data) return # 去掉重復(fù)日期避免接口把最后一天重復(fù)返回 merged pd.concat([old, new]).drop_duplicates(date).sort_values(date) save_daily(symbol, merged)校驗(yàn)點(diǎn)一drop_duplicates(date)必須做否則重復(fù)行會(huì)讓指標(biāo)計(jì)算出現(xiàn)跳變。校驗(yàn)點(diǎn)二合并后檢查merged[date].is_monotonic_increasing如果不是遞增說明排序有問題回測會(huì)錯(cuò)亂。這兩步看起來簡單但漏掉任何一個(gè)后面排查起來就是黑匣子。3. 指標(biāo)與回測層把選股想法變成可驗(yàn)證的數(shù)字3.1 均線、MACD、RSI 三個(gè)指標(biāo)的向量化寫法指標(biāo)計(jì)算最忌諱用 for 循環(huán)逐行算pandas 的向量化寫法快幾十倍。下面把三個(gè)常用指標(biāo)寫成獨(dú)立函數(shù)輸入統(tǒng)一是帶close列的 DataFrame輸出新增指標(biāo)列。import pandas as pd import numpy as np def add_ma(df: pd.DataFrame, windows(5, 20, 60)) - pd.DataFrame: for w in windows: df[fma{w}] df[close].rolling(w).mean() return df def add_macd(df: pd.DataFrame, fast12, slow26, signal9) - pd.DataFrame: ema_fast df[close].ewm(spanfast, adjustFalse).mean() ema_slow df[close].ewm(spanslow, adjustFalse).mean() df[dif] ema_fast - ema_slow df[dea] df[dif].ewm(spansignal, adjustFalse).mean() df[macd] (df[dif] - df[dea]) * 2 return df def add_rsi(df: pd.DataFrame, period14) - pd.DataFrame: delta df[close].diff() gain delta.clip(lower0).rolling(period).mean() loss (-delta.clip(upper0)).rolling(period).mean() rs gain / loss.replace(0, np.nan) df[rsi] 100 - 100 / (1 rs) return df參數(shù)說明均線窗口 5/20/60 是常見短中長組合改窗口只影響信號(hào)敏感度不影響計(jì)算邏輯。MACD 的adjustFalse必須加否則 ewm 會(huì)按整個(gè)序列長度做調(diào)整結(jié)果和通達(dá)信對(duì)不上。RSI 里loss.replace(0, np.nan)是防止除零停牌或一字板時(shí)會(huì)出現(xiàn)這種情況。3.2 一個(gè)最小回測引擎信號(hào)、持倉、收益三段式回測引擎不需要復(fù)雜核心就三件事根據(jù)指標(biāo)生成買賣信號(hào)、按信號(hào)模擬持倉、算累計(jì)收益。下面是一個(gè)單標(biāo)的、全倉進(jìn)出的最小版本。def backtest(df: pd.DataFrame, signal_col: str signal) - pd.DataFrame: df df.copy() df[position] df[signal_col].shift(1).fillna(0) # 次日開盤執(zhí)行 df[ret] df[close].pct_change().fillna(0) df[strategy_ret] df[position] * df[ret] df[equity] (1 df[strategy_ret]).cumprod() return df def make_signal(df: pd.DataFrame) - pd.DataFrame: # 金叉買入死叉賣出 df[signal] 0 df.loc[df[ma5] df[ma20], signal] 1 df.loc[df[ma5] df[ma20], signal] 0 return df邏輯說明shift(1)是關(guān)鍵信號(hào)當(dāng)天收盤后才知道只能次日?qǐng)?zhí)行不加這一行就是未來函數(shù)回測收益會(huì)虛高。position用 0/1 表示空倉滿倉想加倉位管理就把這里改成 0 到 1 之間的浮點(diǎn)數(shù)。equity是凈值曲線最后拿它算最大回撤和年化。3.3 回測結(jié)果里必須看的三個(gè)數(shù)字跑完回測別只看總收益。我一般先看三個(gè)數(shù)年化收益、最大回撤、夏普比率。年化收益高但回撤 60% 的策略實(shí)盤拿不住。夏普低于 1 的策略說明收益是靠承擔(dān)波動(dòng)換來的性價(jià)比不高。def metrics(df: pd.DataFrame) - dict: total df[equity].iloc[-1] - 1 days (df[date].iloc[-1] - df[date].iloc[0]).days annual (1 total) ** (365 / days) - 1 drawdown (df[equity] / df[equity].cummax() - 1).min() sharpe df[strategy_ret].mean() / df[strategy_ret].std() * (252 ** 0.5) return {annual: annual, max_drawdown: drawdown, sharpe: sharpe}參數(shù)上252 是 A 股一年交易日數(shù)用 365 算夏普會(huì)偏低。最大回撤用cummax算邏輯是當(dāng)前凈值相對(duì)歷史最高點(diǎn)的跌幅。這三個(gè)數(shù)字出來策略值不值得繼續(xù)調(diào)心里就有數(shù)了。4. 接口與前端讓分析結(jié)果能被瀏覽器訪問4.1 用 FastAPI 暴露三個(gè)核心接口后端接口不用多三個(gè)就夠股票列表、單只股票指標(biāo)數(shù)據(jù)、回測結(jié)果。FastAPI 自帶文檔調(diào)試方便。from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import pandas as pd app FastAPI() app.add_middleware(CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*]) app.get(/api/stock/{symbol}) def get_stock(symbol: str): df pd.read_parquet(fdata/{symbol}.parquet) df add_ma(add_macd(add_rsi(df))) return df.tail(250).to_dict(orientrecords) app.get(/api/backtest/{symbol}) def get_backtest(symbol: str): df pd.read_parquet(fdata/{symbol}.parquet) df make_signal(add_ma(df)) df backtest(df) return metrics(df)邏輯說明CORS中間件必須加否則前端頁面跨域請(qǐng)求會(huì)被瀏覽器攔。tail(250)只返回最近一年數(shù)據(jù)前端畫圖夠用全量返回會(huì)讓接口變慢。參數(shù)上symbol直接拼進(jìn)文件路徑生產(chǎn)環(huán)境要做白名單校驗(yàn)防止路徑穿越。4.2 前端用 ECharts 畫 K 線和指標(biāo)的最小頁面前端不需要框架也能跑一個(gè) HTML 加 ECharts CDN 就夠。核心是把接口返回的數(shù)組喂給 ECharts 的 candlestick 和 line。!DOCTYPE html html head script srchttps://cdn.jsdelivr.net/npm/echarts5/dist/echarts.min.js/script /head body div idchart stylewidth:100%;height:600px;/div script fetch(http://localhost:8000/api/stock/000001) .then(r r.json()) .then(data { const dates data.map(d d.date); const kline data.map(d [d.open, d.close, d.low, d.high]); const ma5 data.map(d d.ma5); const chart echarts.init(document.getElementById(chart)); chart.setOption({ xAxis: { type: category, data: dates }, yAxis: { scale: true }, series: [ { type: candlestick, data: kline }, { type: line, data: ma5, smooth: true } ] }); }); /script /body /html邏輯說明ECharts 的 candlestick 數(shù)據(jù)順序是[open, close, low, high]順序錯(cuò)了圖會(huì)亂。scale: true讓 Y 軸不強(qiáng)制從 0 開始K 線形態(tài)才看得清。前端頁面直接雙擊打開即可接口地址寫死 localhost部署時(shí)改成實(shí)際域名。5. 避坑與排查這套系統(tǒng)最容易翻車的五個(gè)地方5.1 復(fù)權(quán)沒做對(duì)回測收益全是假的現(xiàn)象回測年化 80%實(shí)盤一跑就虧。原因用了不復(fù)權(quán)價(jià)除權(quán)除息日價(jià)格跳空被當(dāng)成真實(shí)漲跌。解決數(shù)據(jù)層統(tǒng)一用adjustqfq并且回測和指標(biāo)計(jì)算讀同一份復(fù)權(quán)數(shù)據(jù)不要一個(gè)用復(fù)權(quán)一個(gè)用不復(fù)權(quán)。5.2 未來函數(shù)藏在 shift 里現(xiàn)象策略勝率 90%但信號(hào)出現(xiàn)當(dāng)天就買入。原因signal沒做shift(1)用了當(dāng)天收盤價(jià)生成的信號(hào)去算當(dāng)天收益。解決所有信號(hào)列在進(jìn)入回測前統(tǒng)一shift(1)并且檢查position列的第一行是不是 0。5.3 停牌日導(dǎo)致指標(biāo)斷層現(xiàn)象某只股票均線突然跳變。原因停牌期間沒有行情數(shù)據(jù)rolling 窗口跨過了停牌日。解決拉數(shù)據(jù)時(shí)用交易日歷對(duì)齊停牌日填充前收盤價(jià)或者直接在指標(biāo)計(jì)算前dropna掉停牌區(qū)間。5.4 接口返回中文列名導(dǎo)致前端報(bào)錯(cuò)現(xiàn)象前端拿到的數(shù)據(jù)里 key 是「日期」「收盤」ECharts 取不到值。原因數(shù)據(jù)層沒統(tǒng)一列名直接透傳了原始中文列。解決在數(shù)據(jù)層rename成英文接口層只返回英文 key前端按英文 key 取值。5.5 parquet 文件被并發(fā)寫入損壞現(xiàn)象更新數(shù)據(jù)時(shí)程序崩潰parquet 文件讀不出來。原因多個(gè)進(jìn)程同時(shí)寫同一個(gè)文件。解決更新任務(wù)串行執(zhí)行或者先寫臨時(shí)文件再os.replace原子替換。我一般會(huì)在更新函數(shù)里加文件鎖避免定時(shí)任務(wù)和手動(dòng)更新撞車。6. 進(jìn)階技巧用參數(shù)網(wǎng)格快速驗(yàn)證策略穩(wěn)健性單跑一組參數(shù)說明不了問題真正要看的是策略在不同參數(shù)下的表現(xiàn)是否穩(wěn)定。我一般會(huì)做參數(shù)網(wǎng)格把均線窗口、MACD 參數(shù)、RSI 閾值組合起來跑一遍看收益和回撤的分布。import itertools def grid_search(df: pd.DataFrame, ma_windows, rsi_periods): results [] for ma, rsi in itertools.product(ma_windows, rsi_periods): d add_ma(df.copy(), windows(ma, ma * 4)) d add_rsi(d, periodrsi) d[signal] (d[fma{ma}] d[fma{ma*4}]).astype(int) d backtest(d) m metrics(d) results.append({ma: ma, rsi: rsi, **m}) return pd.DataFrame(results).sort_values(sharpe, ascendingFalse)參數(shù)說明ma_windows傳[5, 10, 20]rsi_periods傳[6, 14, 24]組合出 9 組結(jié)果?？磗harpe排序如果前幾名參數(shù)差異很大但夏普都高于 1說明策略穩(wěn)健如果只有某一組特別高大概率是過擬合。這一步跑完再?zèng)Q定要不要上實(shí)盤模擬。我自己的習(xí)慣是任何策略在網(wǎng)格搜索里夏普波動(dòng)超過 50%就直接放棄不浪費(fèi)時(shí)間去調(diào)。這套系統(tǒng)搭起來大概兩天但省下的是后面無數(shù)次重復(fù)造輪子的時(shí)間。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取