簡介這是一套基于Python構(gòu)建的全棧股票分析系統(tǒng)源碼面向金融數(shù)據(jù)分析學習者、量化研究入門者及需要搭建個性化行情工具的開發(fā)者。系統(tǒng)以akshare為核心數(shù)據(jù)接口覆蓋A股、港股、美股等多市場行情并整合pandas、TensorFlow、tornado等工具實現(xiàn)數(shù)據(jù)采集、統(tǒng)計建模與可視化展示的完整鏈路同時支持實時追蹤與歷史回溯。資源包共172個文件約2.18MB包含29個py核心腳本、34個js與19個html/css前端頁面、28個zbak備份文件以及dockerfile、conf、sh等部署配置目錄結(jié)構(gòu)清晰便于按模塊閱讀與二次開發(fā)。已有47人學習下載。代碼經(jīng)過重構(gòu)模塊化程度較高pandas強化了時間序列清洗能力tornado異步架構(gòu)支撐高并發(fā)實時更新TensorFlow用于預(yù)測與風險評估建模適合作為金融分析工具鏈的實踐參考。1. 一套能跑起來的 Python 全棧股票分析系統(tǒng)到底由哪些模塊拼成很多人搜「基于Python的全棧股票分析系統(tǒng)源碼」腦子里想的是下載一個壓縮包、解壓、python app.py然后瀏覽器里就出現(xiàn)一個能看 K 線、能選股、能回測的完整系統(tǒng)?，F(xiàn)實是這類項目從來不是「一個文件跑天下」而是數(shù)據(jù)采集、數(shù)據(jù)存儲、指標計算、策略回測、Web 接口、前端展示這幾塊拼起來的。全棧兩個字重點不在「全」而在「?！埂恳粚佣家塥毩⑴芡ㄔ俅梢粭l鏈路。這套系統(tǒng)適合兩類人一類是剛學完 Python 基礎(chǔ)語法、想找一個真實項目練手的開發(fā)者另一類是手里有選股想法、但缺一套可復(fù)用分析框架的從業(yè)者。它解決的核心問題是把「拉數(shù)據(jù) → 算指標 → 出信號 → 看結(jié)果」這條鏈路固化下來而不是每次都在 Jupyter 里重新寫一遍。下面按模塊拆開講每一步都落到能復(fù)現(xiàn)的命令和代碼上。2. 數(shù)據(jù)層怎么搭從行情接口到本地緩存的完整鏈路2.1 為什么先定數(shù)據(jù)源再寫任何分析代碼股票分析系統(tǒng)的第一道坎不是算法是數(shù)據(jù)。常見做法是先用公開行情接口拉日線數(shù)據(jù)落到本地做緩存后續(xù)所有指標計算和回測都讀本地文件。這樣做的好處是接口掛了不影響你調(diào)試策略回測速度也快一個量級。我一般會選akshare或tushare這類庫做數(shù)據(jù)入口前者免費、覆蓋 A 股較全后者需要 token 但字段規(guī)范。選型時看三個點一是復(fù)權(quán)處理是否內(nèi)置二是歷史數(shù)據(jù)能拉多長三是請求頻率限制。日線級別分析前兩點比第三點重要因為日線數(shù)據(jù)一天只需要更新一次頻率壓力不大。把數(shù)據(jù)源定下來之后目錄結(jié)構(gòu)就要提前規(guī)劃否則后面指標、回測、接口三層會互相污染。# 項目根目錄結(jié)構(gòu)先建好再寫代碼 stock-analysis/ ├── data/ # 本地行情緩存按股票代碼分文件 ├── indicators/ # 指標計算模塊 ├── backtest/ # 回測引擎 ├── api/ # Web 接口層 ├── web/ # 前端頁面 └── config.py # 數(shù)據(jù)源、路徑、參數(shù)集中配置這個結(jié)構(gòu)的關(guān)鍵是data/和config.py分離。數(shù)據(jù)文件不進版本控制配置集中管理換數(shù)據(jù)源時只改config.py一個地方。2.2 用 akshare 拉日線并落地成 parquet 的最小腳本下面這段代碼做三件事拉一只股票的日線、做前復(fù)權(quán)、存成 parquet。parquet 比 CSV 快列式存儲對指標計算友好文件體積也小。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)價 ) # 統(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)價算出來的收益率是錯的。列名統(tǒng)一成英文是為了后面指標模塊不依賴中文列名換數(shù)據(jù)源時只改這一個函數(shù)。參數(shù)上start_date和end_date用字符串傳入akshare 內(nèi)部會解析不用自己轉(zhuǎn) datetime。2.3 增量更新與數(shù)據(jù)校驗的兩個必做檢查全量拉一次之后日常只需要增量更新。增量邏輯是讀本地 parquet 的最后一條日期從那天往后拉。這里有兩個坑一是停牌日會導(dǎo)致日期不連續(xù)二是接口偶爾返回空 DataFrame。所以更新函數(shù)里必須做校驗。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)校驗點一drop_duplicates(date)必須做否則重復(fù)行會讓指標計算出現(xiàn)跳變。校驗點二合并后檢查merged[date].is_monotonic_increasing如果不是遞增說明排序有問題回測會錯亂。這兩步看起來簡單但漏掉任何一個后面排查起來就是黑匣子。3. 指標與回測層把選股想法變成可驗證的數(shù)字3.1 均線、MACD、RSI 三個指標的向量化寫法指標計算最忌諱用 for 循環(huán)逐行算pandas 的向量化寫法快幾十倍。下面把三個常用指標寫成獨立函數(shù)輸入統(tǒng)一是帶close列的 DataFrame輸出新增指標列。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 是常見短中長組合改窗口只影響信號敏感度不影響計算邏輯。MACD 的adjustFalse必須加否則 ewm 會按整個序列長度做調(diào)整結(jié)果和通達信對不上。RSI 里loss.replace(0, np.nan)是防止除零停牌或一字板時會出現(xiàn)這種情況。3.2 一個最小回測引擎信號、持倉、收益三段式回測引擎不需要復(fù)雜核心就三件事根據(jù)指標生成買賣信號、按信號模擬持倉、算累計收益。下面是一個單標的、全倉進出的最小版本。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)鍵信號當天收盤后才知道只能次日執(zhí)行不加這一行就是未來函數(shù)回測收益會虛高。position用 0/1 表示空倉滿倉想加倉位管理就把這里改成 0 到 1 之間的浮點數(shù)。equity是凈值曲線最后拿它算最大回撤和年化。3.3 回測結(jié)果里必須看的三個數(shù)字跑完回測別只看總收益。我一般先看三個數(shù)年化收益、最大回撤、夏普比率。年化收益高但回撤 60% 的策略實盤拿不住。夏普低于 1 的策略說明收益是靠承擔波動換來的性價比不高。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 算夏普會偏低。最大回撤用cummax算邏輯是當前凈值相對歷史最高點的跌幅。這三個數(shù)字出來策略值不值得繼續(xù)調(diào)心里就有數(shù)了。4. 接口與前端讓分析結(jié)果能被瀏覽器訪問4.1 用 FastAPI 暴露三個核心接口后端接口不用多三個就夠股票列表、單只股票指標數(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中間件必須加否則前端頁面跨域請求會被瀏覽器攔。tail(250)只返回最近一年數(shù)據(jù)前端畫圖夠用全量返回會讓接口變慢。參數(shù)上symbol直接拼進文件路徑生產(chǎn)環(huán)境要做白名單校驗防止路徑穿越。4.2 前端用 ECharts 畫 K 線和指標的最小頁面前端不需要框架也能跑一個 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]順序錯了圖會亂。scale: true讓 Y 軸不強制從 0 開始K 線形態(tài)才看得清。前端頁面直接雙擊打開即可接口地址寫死 localhost部署時改成實際域名。5. 避坑與排查這套系統(tǒng)最容易翻車的五個地方5.1 復(fù)權(quán)沒做對回測收益全是假的現(xiàn)象回測年化 80%實盤一跑就虧。原因用了不復(fù)權(quán)價除權(quán)除息日價格跳空被當成真實漲跌。解決數(shù)據(jù)層統(tǒng)一用adjustqfq并且回測和指標計算讀同一份復(fù)權(quán)數(shù)據(jù)不要一個用復(fù)權(quán)一個用不復(fù)權(quán)。5.2 未來函數(shù)藏在 shift 里現(xiàn)象策略勝率 90%但信號出現(xiàn)當天就買入。原因signal沒做shift(1)用了當天收盤價生成的信號去算當天收益。解決所有信號列在進入回測前統(tǒng)一shift(1)并且檢查position列的第一行是不是 0。5.3 停牌日導(dǎo)致指標斷層現(xiàn)象某只股票均線突然跳變。原因停牌期間沒有行情數(shù)據(jù)rolling 窗口跨過了停牌日。解決拉數(shù)據(jù)時用交易日歷對齊停牌日填充前收盤價或者直接在指標計算前dropna掉停牌區(qū)間。5.4 接口返回中文列名導(dǎo)致前端報錯現(xiàn)象前端拿到的數(shù)據(jù)里 key 是「日期」「收盤」ECharts 取不到值。原因數(shù)據(jù)層沒統(tǒng)一列名直接透傳了原始中文列。解決在數(shù)據(jù)層rename成英文接口層只返回英文 key前端按英文 key 取值。5.5 parquet 文件被并發(fā)寫入損壞現(xiàn)象更新數(shù)據(jù)時程序崩潰parquet 文件讀不出來。原因多個進程同時寫同一個文件。解決更新任務(wù)串行執(zhí)行或者先寫臨時文件再os.replace原子替換。我一般會在更新函數(shù)里加文件鎖避免定時任務(wù)和手動更新撞車。6. 進階技巧用參數(shù)網(wǎng)格快速驗證策略穩(wěn)健性單跑一組參數(shù)說明不了問題真正要看的是策略在不同參數(shù)下的表現(xiàn)是否穩(wěn)定。我一般會做參數(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)健如果只有某一組特別高大概率是過擬合。這一步跑完再決定要不要上實盤模擬。我自己的習慣是任何策略在網(wǎng)格搜索里夏普波動超過 50%就直接放棄不浪費時間去調(diào)。這套系統(tǒng)搭起來大概兩天但省下的是后面無數(shù)次重復(fù)造輪子的時間。希望幫到你。本文還有配套的精品資源點擊獲取