AI 量化学习笔记 · 第 3 篇
上一篇还在等券商的书面答复，但研究环境没必要跟着停下来。数据、回测和因子研究都放在研究层，不绑定券商终端，现在就能做。 这篇记录我在 MacBook 上搭建研究环境的过程，并用 510300（沪深 300 ETF）跑通日线数据的拉取、入库、复权检查和画图。完整代码放在 https://github.com/wangyiyang/quant。
两条边界
前 3 个月不买新硬件。一只 ETF 十年的日线只有两千多行，即使扩到几百只 ETF，也只是几十万行。Apple Silicon 足够处理这个量级，先把现有机器用起来。 研究和执行也要分开。研究层只输出目标仓位；Windows、券商终端和下单接口留到第 4—6 个月处理。这样以后换券商或终端，研究代码不用跟着重写。
先把目录搭好
我不想从一个孤零零的 Jupyter Notebook 开始。Notebook 很适合试想法，但内容一多，很快就分不清哪个 cell 才是最终版本。所以第一天我就把目录、依赖和版本控制搭好。
quant/
├── data/ # DuckDB 文件与原始缓存（不进 Git）
├── src/
│ ├── fetch/ # 数据拉取：akshare / baostock
│ ├── db/ # 入库与查询
│ └── viz/ # 画图
├── notebooks/ # 探索性分析，结论必须回流到 src/
├── tests/
├── .env # 密钥与配置（不进 Git）
└── README.md
初始化只有几条命令，一次性做完就不用再想：
mkdir -p quant/{data,src/{fetch,db,viz},notebooks,tests}
cd quant
python3.11 -m venv .venv && source .venv/bin/activate
pip install akshare baostock duckdb pandas numpy plotly pytest python-dotenv
pip freeze > requirements.txt
printf 'data/\n.venv/\n.env\n__pycache__/\n' > .gitignore
git init && git add -A && git commit -m "chore: init quant research env"
Notebook 只负责探索；需要复用的逻辑放进 src/ 并补测试。数据文件和密钥不进 Git。
软件栈
环境用 Python 3.11 和 venv，数据处理与画图选 pandas、numpy、plotly。日线数据量不大，先用熟悉的工具，不为暂时用不到的性能引入 polars。行情由 akshare 拉取，baostock 用来交叉核对。 数据落在 DuckDB。CSV 遇到复权因子更新时需要重写文件，历史数据一多就容易失控；SQLite 也能做，但 DuckDB 的列存和分析查询更适合这类工作负载，而且同样只需要一个本地文件。
跑通 510300 日线
流程是拉取、入库、检查复权，再画图。
拉取
akshare 的东财接口不需要 token。adjust 参数决定返回原始价还是复权价，这里两份都保存：
# src/fetch/akshare_daily.py
import akshare as ak
import pandas as pd
COLS = {
"日期": "date", "开盘": "open", "最高": "high", "最低": "low",
"收盘": "close", "成交量": "volume", "成交额": "amount",
}
def fetch_etf_daily(symbol="510300", start="20130101", end="20260726", adjust="hfq"):
"""adjust: "" 原始价 / "hfq" 后复权"""
df = ak.fund_etf_hist_em(
symbol=symbol, period="daily",
start_date=start, end_date=end, adjust=adjust,
).rename(columns=COLS)
df["date"] = pd.to_datetime(df["date"]).dt.date
df["symbol"] = symbol
df["adjust"] = adjust or "raw"
return (df[["symbol", "date", "open", "high", "low", "close", "volume", "amount", "adjust"]]
.sort_values("date").reset_index(drop=True))
入库
入库必须幂等：同一天重复拉取只能留一行，复权因子变化后也能整段回写。这里用主键定位旧记录，先删再插：
# src/db/store.py
import pathlib
import duckdb
DB_PATH = pathlib.Path("data/quant.duckdb")
SCHEMA = """
create table if not exists etf_daily (
symbol varchar, date date,
open double, high double, low double, close double,
volume bigint, amount double, adjust varchar,
primary key (symbol, date, adjust)
);
"""
def upsert_daily(df):
with duckdb.connect(str(DB_PATH)) as con:
con.execute(SCHEMA)
con.register("incoming", df)
con.execute("""delete from etf_daily e
using incoming i
where e.symbol = i.symbol and e.date = i.date and e.adjust = i.adjust""")
con.execute("insert into etf_daily select * from incoming")
return con.execute("select count(*) from etf_daily").fetchone()[0]
def load_daily(symbol="510300", adjust="hfq"):
with duckdb.connect(str(DB_PATH), read_only=True) as con:
return con.execute("""select date, open, high, low, close, volume
from etf_daily where symbol = ? and adjust = ?
order by date""", [symbol, adjust]).df()
画图
plotly 负责 K 线和成交量副图，结果写成 HTML，直接用浏览器查看：
# src/viz/candles.py
import plotly.graph_objects as go
from plotly.subplots import make_subplots
def plot_candles(df, symbol="510300", out="data/510300.html"):
fig = make_subplots(rows=2, cols=1, shared_xaxes=True,
row_heights=[0.72, 0.28], vertical_spacing=0.03)
fig.add_trace(go.Candlestick(x=df["date"], open=df["open"], high=df["high"],
low=df["low"], close=df["close"], name="price"), row=1, col=1)
fig.add_trace(go.Bar(x=df["date"], y=df["volume"], name="volume"), row=2, col=1)
fig.update_layout(title=f"{symbol} 日线（后复权）", xaxis_rangeslider_visible=False,
height=760, template="plotly_white")
fig.write_html(out)
return out
串起来运行
# scripts/run_pipeline.py
from src.fetch.akshare_daily import fetch_etf_daily
from src.db.store import upsert_daily, load_daily
from src.viz.candles import plot_candles
for adjust in ("", "hfq"):
rows = upsert_daily(fetch_etf_daily(adjust=adjust))
print(f"adjust={adjust or 'raw':>3} -> 库内累计 {rows} 行")
df = load_daily("510300", adjust="hfq")
print(df.tail(3))
print("chart ->", plot_candles(df))
执行方式有个细节：脚本里用了 from src... 的包导入，必须在项目根目录以模块方式运行；直接 python scripts/run_pipeline.py 会把 scripts/ 当成搜索根，报 ModuleNotFoundError: No module named 'src'。
cd quant && source .venv/bin/activate
python -m scripts.run_pipeline
图能正常打开之后，我先检查了复权。 不复权的价格序列会在分红除息日向下跳空，直接拿它计算动量会产生假信号。我的处理是：后复权价格用于研究，原始价格保留下来核对成交。两份数据都在库里后，可以用 SQL 比较同一天的收益率；差值明显不为零的日期，通常就是分红除息日：
-- 找出原始价与后复权价日收益率背离的交易日
with r as (
select date, adjust,
close / lag(close) over (partition by adjust order by date) - 1 as ret
from etf_daily where symbol = '510300'
)
select a.date,
round(a.ret * 100, 3) as raw_pct,
round(b.ret * 100, 3) as hfq_pct
from r a join r b on a.date = b.date
where a.adjust = 'raw' and b.adjust = 'hfq'
and abs(a.ret - b.ret) > 0.002
order by a.date;
这段 SQL 不放进 run_pipeline.py，而是保存为独立脚本 scripts/check_adjust.py。脚本以只读方式连接 data/quant.duckdb，执行上面的查询并打印背离日期；完整实现可以在 GitHub 仓库中查看。 
先运行 run_pipeline，把 raw 和 hfq 两份数据入库，再执行 python -m scripts.check_adjust。临时检查也可以安装 DuckDB CLI（brew install duckdb），然后用 duckdb -readonly data/quant.duckdb 打开。只读模式可以避免与正在写库的脚本争用连接。 
另一个容易混淆的地方是交易日。「过去 20 天」和「过去 20 个交易日」不是一回事，窗口计算必须按交易记录或交易所日历走，不能直接减 20 个自然日。
用第二个数据源核对
免费数据最麻烦的地方，是错误通常不会主动报出来。我用 baostock 再拉一份 510300 日线，与 akshare 的结果按日期对齐，检查收盘价偏差和单边缺失的日期。
# src/fetch/crosscheck.py
import baostock as bs
import pandas as pd
from src.db.store import load_daily
def fetch_baostock(code="sh.510300", start="2013-01-01", end="2026-07-26"):
bs.login()
try:
rs = bs.query_history_k_data_plus(
code, "date,close,volume", start_date=start, end_date=end,
frequency="d", adjustflag="3", # 3 = 不复权，与 raw 对齐
)
if rs.error_code != "0":
raise RuntimeError(f"baostock 查询失败：{rs.error_code} {rs.error_msg}")
rows = []
while rs.error_code == "0" and rs.next():
rows.append(rs.get_row_data())
print(f"baostock 获取 {len(rows)} 行，最后一天：{rows[-1][0] if rows else '无数据'}")
finally:
bs.logout()
df = pd.DataFrame(rows, columns=["date", "close", "volume"])
df["date"] = pd.to_datetime(df["date"])
return df.astype({"close": float, "volume": float})
def crosscheck(tol=0.005):
a = load_daily("510300", adjust="raw").rename(
columns={"close": "close_ak", "volume": "vol_ak"}
)
b = fetch_baostock().rename(
columns={"close": "close_bs", "volume": "vol_bs"}
)
a["date"] = pd.to_datetime(a["date"])
b["date"] = pd.to_datetime(b["date"])
m = a.merge(b, on="date", how="outer", indicator=True)
only = m[m["_merge"] != "both"][["date", "_merge"]]
both = m[m["_merge"] == "both"].copy()
both["close_diff"] = (
(both["close_ak"] - both["close_bs"]).abs() / both["close_bs"]
)
bad = both[both["close_diff"] > tol][
["date", "close_ak", "close_bs", "close_diff"]
]
print(f"共同交易日 {len(both)}，收盘价一致率 {1 - len(bad) / len(both):.4%}")
print(f"仅单边存在的日期 {len(only)} 天")
return bad.sort_values("date"), only
注意 adjustflag="3" 是不复权，必须和库里的 raw 那份比，拿复权价去对不复权价只会得到一堆假差异。 
结果有两个数字值得记下来。两边共有 134 个交易日重叠，收盘价一致率为 100%，没有发现超过阈值的差异。与此同时，3157 个日期全部是 left_only，也就是 akshare 有数据、baostock 没有。baostock 这次只返回了 134 行，最后一天是 2026-07-24。 抓取过程没有报错，因此这里能确认的是：baostock 对本次 510300 查询只返回了较短的一段历史。重叠区间可以完成双源核对，更早的数据仍然只有 akshare 一个来源。后面做长期回测时，我会把这段数据边界写进结果说明，不把「部分区间验证通过」说成「全部历史都可靠」。 研究环境先停在这里。下一步开始写回测器，先保证每一笔交易都能复现。
本系列不构成投资建议。终端权限、资金门槛及程序化交易要求以券商的最新书面答复为准。
文档信息
- 本文作者：王翊仰
- 本文链接：https://www.wangyiyang.cc/2026/07/27/macbook/
- 版权声明：自由转载-非商用-非衍生-保持署名（创意共享3.0许可证）