企业官网建设流程全解析

简介：这是一套基于Python构建的股票量化交易系统，面向对量化投资、程序化交易和机器学习感兴趣的开发者与投资者。资源完整呈现阿布（Abu）量化框架的设计思路与实现，涵盖数据获取、策略回测、资金管理、实时交易等核心模块，并结合机器学习模型用于趋势预测与情绪分析，适合从入门到进阶的量化研究者参考实践。压缩包共包含383个文件，约59.16MB，以248个Python源码文件为主体，辅以51个Jupyter Notebook演示文档，便于逐步理解策略逻辑；同时附带多张PNG/GIF运行效果图、CSV行情样本及模型配置文件，覆盖从代码、数据到可视化展示的完整闭环。目录结构清晰，便于按模块检索学习。目前已有1636人浏览学习。通过研读源码与案例，读者可以掌握量化回测框架的搭建方法、策略评价与参数优化技巧，并了解如何将机器学习引入交易决策，是一份兼具理论深度与实操价值的参考资料。

1. 从手工选股到 Python 量化：阿布股票系统先把数据闭环讲透

做股票研究的都知道，手工选股的最大问题不是信息不够，而是信息来了之后你处理不过来。几十个指标、上千只股票、日线周线月线叠加在一起，靠 Excel 和肉眼盯盘，别说做策略迭代，光是每天更新数据就要耗掉一两个小时。这个名为“阿布股票量化系统”的资源，核心价值就是把“数据获取 → 特征计算 → 模型预测 → 回测验证”这一整条链路用 Python 串起来，并且每一个环节都留了可改的接口。它不是那种给你一个黑匣子、跑完出一张收益曲线的演示 Demo，而是把数据清洗、因子计算、模型训练和回测逻辑全部摊开，适合那些已经会 Python 基础语法、想认真落地一套机器学习选股流程的开发者。整套系统面向 A 股日线级别，以机器学习为主干，尤其适合你手里已经有数据源、但始终没搭好工程框架的中间状态。下面我从数据层开始，一层层拆这个资源实际用起来的样子。

2. 数据获取与预处理：先把日线数据洗成可用的 DataFrame

2.1 数据源接入与本地存储结构

任何量化系统第一步都是数据。这个资源在数据层做得比较干脆：它不绑定某个固定数据源，而是把数据拉取和本地存储分开设计。我一般用免费的历史行情接口拉日线数据，同时兼容 tushare 的本地缓存模式，拉下来的数据统一落成 CSV 文件，目录结构按“市场 / 股票代码 / 日线.csv”组织。

import pandas as pd import os from datetime import datetime def fetch_daily(symbol: str, start: str, end: str, data_dir: str = "./stock_data"): """ 拉取单只股票的日线数据并落盘为 CSV :param symbol: 股票代码，如 '600001' :param start: 开始日期 '20200101' :param end: 结束日期 '20241231' :param data_dir: 本地数据根目录 :return: 日线 DataFrame """ file_path = os.path.join(data_dir, f"{symbol}_daily.csv") if os.path.exists(file_path): # 增量更新：只拉取本地最后日期之后的数据 df_existing = pd.read_csv(file_path, parse_dates=["date"]) last_date = df_existing["date"].max() df_new = fetch_from_api(symbol, last_date.strftime("%Y%m%d"), end) df = pd.concat([df_existing, df_new]).drop_duplicates(subset=["date"]) else: df = fetch_from_api(symbol, start, end) df["symbol"] = symbol df.to_csv(file_path, index=False) return df

这里fetch_from_api是对外行情的封装，实际运行时替换成你自己有权限的数据源即可。重点看两个设计：一是增量更新，每次只拉上次存储日期之后的数据，避免全量重拉；二是落盘为 CSV 而不是直接存内存，这样回测阶段读数据可以反复用，不必每次启动都重新请求接口。参数start和end都用字符串形式传入，在函数内部统一转成时间对象，避免时区问题。

2.2 复权、停牌与对齐：三个绕不开的预处理步骤

数据拿到手只是起点。这个资源在预处理里做了三件事，我建议你照着抄：前复权、停牌过滤、交易日对齐。

前复权最容易被新手忽略。股票分红送股之后价格会出现跳空，不复权的话，技术指标和涨跌幅会被扭曲。系统里默认用前复权价计算指标，逻辑很简单：

def adjust_price(df: pd.DataFrame, method: str = "qfq") -> pd.DataFrame: """ 前复权处理：以最新价格为基准，回溯调整历史价格 """ df = df.sort_values("date").copy() if method == "qfq": factor = (df["close"] / df["pre_close"]).fillna(1) cum_factor = (factor.iloc[::-1].cumprod()).iloc[::-1] df["close_adj"] = df["close"] / cum_factor df["open_adj"] = df["open"] / cum_factor df["high_adj"] = df["high"] / cum_factor df["low_adj"] = df["low"] / cum_factor return df

pre_close是前收盘价，用它算出的单日价格变动因子做累计乘积，就是前复权系数。这个实现里有个细节：factor的缺失值要填充为 1，否则停牌日的复权因子会变成 NaN，连锁污染后面所有指标。停牌日的数据本身也有讲究，很多数据源的停牌日会返回 0 成交量或者干脆不返回记录。这个资源的做法是：保留停牌期间的记录但把volume置 0，并在特征计算时用前值填充价格列。

交易日对齐解决的是多股票拼接时日期不一致的问题。A 股有统一交易日历，但停牌的股票会在某个交易日缺数据。我处理时习惯先取全市场交易日的并集，再用 reindex 把每只股票对齐到这个日历上：

all_dates = market_calendar # 全市场交易日列表 df_aligned = df.set_index("date").reindex(all_dates).ffill()

这里ffill()是前向填充，停牌日沿用上一交易日的价格。如果你做的是横截面选股模型，这一步尤其关键——否则不同股票的特征矩阵行数不一样，模型训练时没法 concat。

3. 因子计算与特征工程：构建机器学习能吃的“X 矩阵”

3.1 基础行情因子与技术指标因子

机器学习选股和传统量化最大的差别在于：传统方法靠少数几个逻辑明确的因子做筛选，而机器学习需要大量特征让模型自己找规律。这个资源里内置了大约 30 个因子，分为三类：基础行情因子、技术指标因子、统计特征因子。

基础行情因子就是收益率、振幅、换手率这些直接由 OHLCV 派生出来的值。技术指标因子在这里占了重头，包括 MA 均线偏离度、RSI 相对强弱、MACD 柱值、布林带位置。指标计算我推荐统一封装，不要东一块西一块写在策略里：

def compute_technical_factors(df: pd.DataFrame) -> pd.DataFrame: """ 计算常用技术指标因子并拼接到原始数据 """ result = df.copy() # MA 均线偏离：当前价格相对 20 日均线的偏离百分比 result["ma5"] = result["close"].rolling(5).mean() result["ma20"] = result["close"].rolling(20).mean() result["ma_bias"] = (result["close"] - result["ma20"]) / result["ma20"] # RSI：14 日相对强弱 delta = result["close"].diff() gain = delta.clip(lower=0).rolling(14).mean() loss = (-delta.clip(upper=0)).rolling(14).mean() result["rsi_14"] = 100 - (100 / (1 + gain / loss)) # MACD 柱值：DIF 与 DEA 之差 ema12 = result["close"].ewm(span=12, adjust=False).mean() ema26 = result["close"].ewm(span=26, adjust=False).mean() result["macd_dif"] = ema12 - ema26 result["macd_dea"] = result["macd_dif"].ewm(span=9, adjust=False).mean() result["macd_hist"] = result["macd_dif"] - result["macd_dea"] return result

这里rolling(5).mean()是滚动均值，ewm是指数加权移动平均，adjust=False保证用递推公式计算而不是加权全历史。MACD 参数用的是 12、26、9 这组经典设置，实际调参时你会发现，A 股日线上把 12/26 改成 10/22 在某些年份会有明显差异，这个后面避坑部分细说。因子计算完不是直接丢给模型，还要做一步：滞后处理。所有因子在 T 日算出，但预测的是 T+1 日的收益，所以特征矩阵里X的第 i 行对应的是因子在 i 日收盘后的值，y则是 i+1 日的收益率，两者在时间上不能错位。

3.2 特征矩阵拼接与样本权重

单只股票算完因子，接下来要做全市场拼接。这个资源给了一个很实用的做法：用透视表把所有股票的同一因子拼成一个大矩阵，shape 是「交易日 × 股票数」，然后每个交易日取所有股票在该日的因子值作为一行样本。

def build_feature_matrix(factor_dict: dict, target_col: str = "next_ret"): """ factor_dict: {股票代码: 因子DataFrame} 返回 X（特征矩阵）和 y（目标变量） """ frames = [] for symbol, df in factor_dict.items(): feat = df[["ma_bias", "rsi_14", "macd_hist"]].copy() feat["symbol"] = symbol frames.append(feat) X_raw = pd.concat(frames) # 目标变量：次日收益率 X_raw[target_col] = X_raw.groupby("symbol")["close"].pct_change().shift(-1) X_raw = X_raw.dropna() # 特征列和标签列分离 y = X_raw[target_col] X = X_raw.drop(columns=[target_col]) return X, y

这个实现里藏了一个关键点：groupby("symbol")["close"].pct_change().shift(-1)是在每只股票内部计算次日收益率，shift(-1)表示把 T+1 日的收益平移到 T 日，对齐因子值。注意dropna()这一步会把最后一天的样本删掉，因为最后一天没有次日收益，这是合理的。实际运行中你会发现样本量非常惊人——全市场 5000 只股票、500 个交易日，就是 250 万条样本。这个量级喂给 XGBoost 内存勉强够，但训练会很慢。我一般会加一个样本筛选：只保留当日成交额排名前 800 的股票，这样既能减少噪音股票，又能把训练时间控制在可接受范围。成交额过滤还有一个好处，就是避开那些流动性极差、容易被资金操控的小票。

4. 机器学习预测与信号生成：选模型、定标签、控过拟合

4.1 标签构造与模型选择

机器学习在选股里的核心任务，其实就是预测“明天哪批股票涨的概率更大”。这个资源把问题定义为二分类：以次日收益是否超过某个阈值（比如 1%）为正样本，否则为负样本。阈值的设定直接影响样本平衡度，1% 在震荡市里可能只有不到 20% 的正样本，这时候模型会倾向于全预测负样本，准确率虚高但毫无意义。

模型选择上，资源默认用的是 XGBoost，理由是它对表格数据的拟合能力强、训练速度快、自带正则化项。我在实际跑实验时，会把 LightGBM 也加进来做对比，两者在 A 股日线数据上的表现差异不算大，但 LightGBM 在样本量超过百万时训练速度快出不少。核心训练逻辑可以这样组织：

import xgboost as xgb from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score def train_predict_model(X: pd.DataFrame, y: pd.Series): """ 训练 XGBoost 分类模型，返回模型和特征重要性 """ # 按时间切分，避免随机打乱引入未来数据 split_idx = int(len(X) * 0.8) X_train, X_valid = X.iloc[:split_idx], X.iloc[split_idx:] y_train, y_valid = y.iloc[:split_idx], y.iloc[split_idx:] model = xgb.XGBClassifier( n_estimators=200, max_depth=5, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, reg_lambda=1.0, eval_metric="logloss", early_stopping_rounds=20 ) model.fit(X_train, y_train, eval_set=[(X_valid, y_valid)], verbose=False) return model

这里最容易被忽略的是切分方式。我见过不少人直接用train_test_split(random_state=42)，这等于让模型看到了未来数据——训练集里中间某一天的样本，验证集里恰好是它后面一天的数据，相关性极强，验证集准确率虚高到 80% 都不稀奇。这个资源用的是按位置顺序切分，前 80% 时间训练、后 20% 时间验证，这是时间序列建模的基本盘。eval_metric="logloss"表示用对数损失作为早停指标，当验证集 loss 连续 20 轮不下降就停止训练，相当于一个自动的过拟合刹车。

4.2 从概率到持仓信号：加入行业过滤和置信度门槛

模型输出的predict_proba是概率值，但不是概率高就无脑买入。这个资源里设计了一个信号生成层的过滤逻辑：先取全市场预测概率排名前 5% 的股票，再剔除 ST、上市不满 60 天的新股，最后按行业做分散——单一行业持仓占比不超过 20%。

这个过滤逻辑我是强烈建议保留的。机器学习模型很容易学到一个 bug：某个行业因为几只龙头股连续暴涨，模型给整个行业打高分，结果持仓全挤在一起，回测收益曲线看着陡，实盘一次行业回调就全部吐回去。行业分散是对模型盲区的强制校正，它不是策略的一部分，而是风控的一部分。

def generate_signals(prob_df: pd.DataFrame, top_ratio: float = 0.05) -> pd.DataFrame: """ 根据预测概率生成最终持仓信号 prob_df: 包含 ['symbol', 'prob', 'industry'] 列的 DataFrame """ # 概率排名取前 top_ratio n_select = int(len(prob_df) * top_ratio) candidates = prob_df.nlargest(n_select, "prob").copy() # ST 股与次新股过滤 candidates = candidates[ ~candidates["symbol"].str.contains("ST") ] # 行业持仓限制：每行业最多 20% max_per_industry = int(len(candidates) * 0.2) + 1 selected = candidates.groupby("industry").head(max_per_industry) return selected[["symbol", "prob", "industry"]]

nlargest(n_select, "prob")取概率最高的前 5%，groupby("industry").head()在每组里取前 N 条，实现行业分散。如果你拿到的数据源没有行业字段，也可以用股票代码前两位粗略替代——A 股的代码段本身和板块有对应关系，虽然粗糙但有总比没有强。生成信号之后，下一步就是把信号和历史行情组合起来做回测，看看这套系统到底能不能赚钱。

5. 回测引擎与四个坑盘点：别让净值曲线骗了你

5.1 回测核心逻辑：撮合规则与手续费设计

信号生成之后再漂亮，也要过回测这一关。这个资源的回测引擎设计得比较轻量，没有事件驱动那么复杂，但核心的撮合规则没省：信号在 T 日收盘后产生，T+1 日开盘买入，持有 N 日后按收盘价卖出。这个设计避免了未来函数——所有决策用的都是 T 日及之前的数据，交易发生在 T+1 日。

回测代码的核心逻辑如下：

def run_backtest(signals: pd.DataFrame, price_data: dict, hold_days: int = 5): """ 简单回测：T+1 开盘买入，持有 hold_days 天后卖出 """ records = [] for _, sig in signals.iterrows(): symbol = sig["symbol"] buy_date = sig["signal_date"] + pd.Timedelta(days=1) df = price_data[symbol] # 找到买入日和卖出日 buy_bar = df[df["date"] >= buy_date].iloc[0] sell_date = buy_date + pd.Timedelta(days=hold_days) sell_bar = df[df["date"] >= sell_date].iloc[0] # 扣除手续费：买入万2.5，卖出万2.5+印花税千1 buy_cost = buy_bar["open"] * (1 + 0.00025) sell_income = sell_bar["close"] * (1 - 0.00025 - 0.001) ret = (sell_income - buy_cost) / buy_cost records.append({"symbol": symbol, "return": ret}) result = pd.DataFrame(records) # 等权组合收益 return result["return"].mean(), result["return"].sum()

注意买入价用的是buy_bar["open"]，卖出价用的是sell_bar["close"]，分别对应 T+1 开盘成交和持有期结束收盘成交。手续费这块容易漏，印花税只收卖出、费率千一，佣金双边各万 2.5。这个资源里的回测是等权组合——每笔信号投入相同资金，虽然简单，但比收益率加权更稳健，不至于被一两个极端收益的单子带偏。跑出来的结果里return.sum()是累计收益，return.mean()是单笔平均收益，后者更适合评估信号质量。

5.2 避坑清单：回测常见的四个坑

第一个坑：前视偏差。现象是回测收益曲线漂亮得惊人，年化超过 50%，但信号日期和买入日期是同一天。原因大多是在生成信号时用了当天收盘之后才知道的数据，比如把当天的close同时用于因子计算和成交判断。解决办法是严格分离信息时间和交易时间：T 日收盘算因子、生成信号，T+1 日才允许成交。这个资源里信号生成和回测是分开两个模块执行的，就是强行物理隔离这两步。

第二个坑：涨跌停无法成交。回测里如果信号股票次日一字涨停，开盘价根本买不进。系统里默认加了一个过滤：买入日开盘涨幅超过 7% 的股票直接跳过。不加这个过滤，回测收益会被虚高 3 到 5 个百分点，实盘全部打脸。

第三个坑：停牌期卖出失败。持有期内股票停牌，卖出信号执行时没有价格。这个资源里的做法是顺延到下一个有交易记录的日期，但要注意，顺延期间股票的涨跌已经算进收益里了，你需要决定这个收益算不算你的。我个人的习惯是算，因为停牌本身是持仓风险的一部分。

第四个坑：数据源复权不一致。如果你用的数据源默认是不复权价格，而特征计算却基于前复权价，就会出现信号和回测对不上的情况。最常见的现象是回测曲线在除权日出现断崖。解决方法是统一在数据接入层强制前复权，回测和因子计算都用同一份复权后的数据，不要各拉各的。

这四关过完，净值曲线才有参考价值。但光看到一条曲线还不够，你还需要确认模型不是只在某一段行情里有效。

6. 滚动窗口与参数寻优：让模型在时间维度上“持续体检”

最后一个实用技巧：滚动训练。静态训练一次、然后拿未来数据验证，只能证明模型在过去可行，无法保证下个月还行。A 股市场风格切换快，因子有效性经常三个月就衰减一轮。我一般会把训练改成滚动模式：每 60 个交易日用最近 250 天的数据重新训练一次模型，然后用未来 20 个交易日做模拟验证。

def rolling_train_eval(X, y, train_window=250, retrain_every=60): """滚动训练：窗口内训练、窗口外预测""" results = [] start = 0 while start + train_window < len(X): train_end = start + train_window X_train, y_train = X.iloc[start:train_end], y.iloc[start:train_end] # 未来 20 日验证 test_end = min(train_end + 20, len(X)) X_test, y_test = X.iloc[train_end:test_end], y.iloc[train_end:test_end] model = xgb.XGBClassifier(n_estimators=150, max_depth=4, learning_rate=0.05) model.fit(X_train, y_train) pred_prob = model.predict_proba(X_test)[:, 1] # 记录每段验证期的 AUC from sklearn.metrics import roc_auc_score auc = roc_auc_score(y_test, pred_prob) results.append({"period": train_end, "auc": auc}) start += retrain_every return pd.DataFrame(results)

跑完这个函数你会得到一条 AUC 曲线，如果曲线在某个时点之后持续下滑，说明模型的特征已经开始失效，这时候要做的是回到因子层，而不是急着调模型参数。网格搜索参数也有技巧，不要直接对全量数据跑 GridSearchCV，那会让你在 250 万样本上等上好几个小时。先固定n_estimators=200，只搜max_depth和learning_rate，用上面滚动验证的 AUC 作为评分标准，每组参数都用同一段验证期数据，才能保证可比性。

这套系统真正难的地方从来不是写代码，而是对每一段数据的信任程度。从那以后我每次跑完回测，都会强制自己走一遍“信号时间是否早于成交时间、复权口径是否统一、涨停股票是否过滤”这三条检查，再去看那条净值曲线。希望这个拆解能帮你在量化这条路上少走几步弯路，也少踩几个我已经替你踩过的坑。

本文还有配套的精品资源，点击获取