02 · 数据获取实战

量化研究的地基。本篇覆盖 A 股、期货、加密货币三个市场的取数实战：每段代码都能直接运行，之后的数据清洗、增量存储、质量检查把「数据」从「能拉到」变成「能信任」。
免责声明：本站全部内容仅用于学习与研究，不构成任何投资建议。市场有风险，投资需谨慎。

一、A 股数据：AKShare 与 Tushare

1.1 AKShare：免费、零门槛

AKShare 聚合了多家公开网站的数据，优点是免费、开箱即用（示例代码仅教学用，实盘风险自负）：

python

import akshare as ak

# 获取贵州茅台日线（前复权），符号/字段以最新 akshare 文档为准
df = ak.stock_zh_a_hist(
    symbol="600519",
    period="daily",
    start_date="20200101",
    end_date="20241231",
    adjust="qfq",          # 前复权，见第四节复权说明
)

df.to_csv("data/600519_daily_qfq.csv", index=False, encoding="utf-8-sig")
print(df.head())
print(df.tail())

财务数据同样一行可取（示例代码仅教学用）：

python

import akshare as ak

# 利润表/主要财务指标（接口名可能随版本变动，以文档为准）
fin = ak.stock_financial_abstract(symbol="600519")
print(fin.head())

⚠️ AKShare 的坑

它抓的是网页源，上游页面改版接口就会变，字段名也不稳定。使用时固定 akshare 版本（写进 requirements.txt），接口报错先去 GitHub 看更新说明。

1.2 Tushare Pro：更规范，需要积分

python

import tushare as ts

ts.set_token("你的token")            # tushare.pro 注册后获取
pro = ts.pro_api()

# 日线（不复权），ts_code 格式：代码.市场
df = pro.daily(ts_code="600519.SH", start_date="20200101", end_date="20241231")
df.to_csv("data/600519_daily_tushare.csv", index=False)
print(df.head())

对比	AKShare	Tushare Pro
成本	免费	免费（积分制，高级接口需积分）
数据规范性	一般，字段常变	好，字段有文档
财务数据	有，接口分散	完整（利润表/资产负债表/指标）
稳定性	依赖网页源	较稳

💡 用法建议

取数阶段两个源都跑一遍存成 CSV，研究阶段用 AKShare，发现可疑数据用 Tushare 交叉验证。

二、期货数据：主力合约 K 线

期货合约会到期，散户研究通常用「主力合约」——持仓量最大的那个合约（示例代码仅教学用，接口以最新 akshare 文档为准）：

python

import akshare as ak

# RB0 = 螺纹钢主力连续（符号规则：品种代码 + 0 表示主力），周期可换
df = ak.futures_main_sina(symbol="RB0", start_date="20230101", end_date="20241231")
print(df.columns)
df.to_csv("data/rb_main_daily.csv", index=False, encoding="utf-8-sig")

# 若要单一合约（如 2025 年 1 月交割的螺纹钢 2501 合约）
df_single = ak.futures_zh_daily_sina(symbol="RB2501")
print(df_single.head())

⚠️ 主力合约的注意事项

主力连续由不同合约拼接而成，换月那天会有价格跳空（合约**价差**），回测前要么用复权处理，要么把换月当作一次移仓成本计入；
「连续合约」有按持仓量、按成交量、按指数（所有合约加权）三种算法，别混用；
想拉日线历史看前复权与后复权处理：期货的跳空是真实的**展期**成本，处理方式比股票更讲究，本篇先按「记录换月点、计入移仓成本」处理。

三、加密货币数据：Binance API

加密市场数据在交易所官方 REST 接口，不限量、字段规范，是个人量化的最佳起点（示例代码仅教学用，实盘风险自负）：

python

import time
import requests
import pandas as pd

BASE_URL = "https://api.binance.com/api/v3"   # 现货；合约用 fapi.binance.com，以官方文档为准

def fetch_klines(symbol, interval, start_ms, end_ms, limit=1000):
    """拉取一段 klines，注意限频：Binance 按权重计费，见下文"""
    params = {
        "symbol": symbol,
        "interval": interval,          # 1m/1h/1d/1w...
        "startTime": start_ms,
        "endTime": end_ms,
        "limit": limit,                # 单次最多 1000 根
    }
    r = requests.get(f"{BASE_URL}/klines", params=params, timeout=15)
    r.raise_for_status()
    # 观察剩余权重，贴近上限时自动降速
    used = int(r.headers.get("x-mbx-used-weight-1m", 0))
    if used > 2000:                    # 默认限频 6000/分钟，预留余量
        time.sleep(30)
    time.sleep(0.1)                    # 基础限速
    return r.json()

def klines_to_df(raw):
    cols = ["open_time", "open", "high", "low", "close", "volume",
            "close_time", "quote_vol", "trades", "taker_buy_base",
            "taker_buy_quote", "ignore"]
    df = pd.DataFrame(raw, columns=cols)
    for c in ["open", "high", "low", "close", "volume"]:
        df[c] = df[c].astype(float)
    df["open_time"] = pd.to_datetime(df["open_time"], unit="ms", utc=True)
    return df[["open_time", "open", "high", "low", "close", "volume"]]

# 按时间分页拉 1 小时线，每次 1000 根
start = int(pd.Timestamp("2024-01-01", tz="UTC").timestamp() * 1000)
end = start + 1000 * 3600 * 1000       # 1000 根 1h K 线
raw = fetch_klines("BTCUSDT", "1h", start, end)
df = klines_to_df(raw)
df.to_csv("data/btcusdt_1h.csv", index=False)
print(df.head())

限频与权重：Binance 是**权重制**——不同接口消耗不同权重，x-mbx-used-weight-1m 头会告诉你当前每分钟已用权重。拉历史数据时注意：

权重不是「次数」，是「成本」；klines 单次请求权重随 limit 增大（以官方文档为准）；
被 429/418 限频后要指数退避重试（等 5s → 30s → 120s），不要猛打；
用 time.sleep 留余量 + 头信息监控，是个人脚本的标准做法。

四、数据清洗：缺失值、停牌日与复权

4.1 缺失值

情况	处理
停牌日	A 股当日无交易数据：日线直接缺行，`ffill` 前一行会污染成交量，通常保留 NaN 由策略决定
接口偶发漏数	先用相邻数据 `ffill` 补全价格，再检查成交量是否为 0 判定停牌
加密 24h 交易	正常无缺失；缺口通常意味着数据源问题，直接删段重拉

python

import pandas as pd

df = pd.read_csv("data/600519_daily_qfq.csv", parse_dates=["日期"])

# 统一列名，便于后续处理（AKShare 中文列名）
df = df.rename(columns={"日期": "date", "开盘": "open", "收盘": "close",
                        "最高": "high", "最低": "low", "成交量": "volume"})
df["volume"] = df["volume"].astype(float)

# 成交量几乎为 0 → 当日停牌或异常，价格保留、标记出来
df.loc[df["volume"] < 1, "halted"] = True
df.loc[df["volume"] >= 1, "halted"] = False
print(df[df["halted"] == True].head())   # 确认停牌日数量合理

4.2 复权：前复权 / 后复权 / 不复权

类型	做法	特点	用在哪
前复权（qfq）	以当前价格为准，把历史价格按分红送股向下调整	历史价格随最新分红不断变化	画 K 线、看形态、算指标最常用
后复权（hfq）	以上市首日为基准，把后续价格向上调整	历史价格固定，不会随分红变化	长期收益率计算、回测必须用
不复权	原始成交价	会有除权跳空，指标被虚假「砸坑/顶」	只做短周期（除权日附近无持仓）可用

💡 回测为什么优先后复权

回测用的是「历史价格」，前复权的历史价格会随你拉的截止日期变化——同一天拉不同日期，同一段历史价格不同，结果不可复现。以固定截止日为准时两者一致，但后复权更严谨。

python

# 获取后复权数据用于回测（字段/参数以 akshare 最新文档为准）
df_hfq = ak.stock_zh_a_hist(symbol="600519", period="daily",
                            start_date="20200101", end_date="20241231",
                            adjust="hfq")
df_hfq.to_csv("data/600519_daily_hfq.csv", index=False)

五、增量更新与存储

5.1 每天增量拉取

个人行情数据量小，策略是「每天跑一次脚本」：以本地已有最大日期为起点拉新增（示例代码仅教学用）：

python

import datetime as dt
import pandas as pd
import akshare as ak

def incremental_update(symbol="600519", path="data/600519_daily_qfq.csv"):
    try:
        old = pd.read_csv(path, parse_dates=["日期"])
        last = old["日期"].max()
        start = (last + dt.timedelta(days=1)).strftime("%Y%m%d")
    except FileNotFoundError:
        old, start = None, "20200101"

    new = ak.stock_zh_a_hist(symbol=symbol, period="daily",
                             start_date=start, end_date=dt.date.today().strftime("%Y%m%d"),
                             adjust="qfq")
    if new.empty:
        return
    df = pd.concat([old, new], ignore_index=True).drop_duplicates(subset="日期")
    df.to_csv(path, index=False, encoding="utf-8-sig")

incremental_update()

⚠️ 增量更新必须去重

接口偶尔会重发已更新过的日期，drop_duplicates 按日期去重是增量更新的保底动作。

5.2 存储选型：CSV vs SQLite vs 数据库

方案	适合	优缺点
CSV（单文件按标的/周期分）	日线研究起步	简单、肉眼可查；跨文件合并要自己写
SQLite（标准库 sqlite3）	品种多了之后	免安装、支持 SQL 去重/查询，个人最优解
PostgreSQL/MySQL	多机协作、生产级	个人单机用不上，运维成本高

python

import sqlite3

con = sqlite3.connect("data/market.db")
# 建表：open_time 为主键，重复写入时跳过（INSERT OR IGNORE）
con.execute("""CREATE TABLE IF NOT EXISTS kline_1h (
    open_time TEXT PRIMARY KEY, open REAL, high REAL,
    low REAL, close REAL, volume REAL)""")
df.to_sql("kline_1h", con, if_exists="append", index=False)  # 教学简化：生产用 INSERT OR IGNORE
con.commit()

💡 个人经验

单标的 CSV + 多标的 SQLite 是最舒服的组合。CSV 给人看、给 git 外的备份；SQLite 给脚本查。

🛑 数据不是拉完就信

数据错误是回测失真最重要的隐性来源：复权方式错误、时区错位、接口漏数、未来数据混入，任何一个都会让回测结果严重偏离真实。 多源交叉验证 + 与行情软件对数字，是数据进入回测前的必经检查。

六、数据质量检查

数据不是拉完就信，跑回测前至少过三关：

数值合理性：high >= low、volume >= 0、无 NaN 价格（df.isna().sum() 一眼扫）。
与行情软件对数字：随机抽 3-5 个日期，和你手机行情软件的手工 K 线对开盘/收盘/成交量，比对不上说明接口源有问题。
区间抽样：看有分红送股的除权日附近，价格是否按复权规则正确跳变；加密数据检查相邻 K 线 open_time 是否连续（无缺根）。

python

# 快速体检脚本
assert (df["high"] >= df["low"]).all(), "high < low 异常"
assert (df["close"].dropna() > 0).all(), "价格非正"
assert df["date"].is_unique, "存在重复日期"
assert df["date"].is_monotonic_increasing, "日期未按升序"
print("基础体检通过")

七、合法合规提醒

免费接口：AKShare/Tushare/Binance 等都有限频与服务条款。爬取频率过高会被封 IP/封账号，商用需看各平台授权条款（Tushare 免费数据不可商用分发，以最新条款为准）。
数据版权：聚宽/米筐/Wind 等付费平台的数据仅限平台内使用，不能下载导出另作他用，这是合同约束，不只是道德问题。
加密接口：Binance/OKX 部分功能存在地区限制，合规问题以你所在司法辖区规定为准；测试/生产使用前阅读官方条款。

八、下一步

数据干净了，进入 03-第一个回测：把这份数据跑成人生第一个带手续费和**滑点**的回测。

⚠️ 风险提示

数据错误是回测失真最重要的隐性来源：复权方式错误、时区错位、接口漏数、未来数据混入，任何一个都会让回测结果严重偏离真实。请务必以「多源交叉验证 + 与行情软件对数字」作为数据进入回测前的必经检查，并将数据获取时间与数据版本记录在案。本文示例代码仅教学用，实盘风险自负；免费数据接口的条款与限频以各平台最新规定为准。

02 · 数据获取实战 ​

一、A 股数据：AKShare 与 Tushare ​

1.1 AKShare：免费、零门槛 ​

1.2 Tushare Pro：更规范，需要积分 ​

二、期货数据：主力合约 K 线 ​

三、加密货币数据：Binance API ​

四、数据清洗：缺失值、停牌日与复权 ​

4.1 缺失值 ​

4.2 复权：前复权 / 后复权 / 不复权 ​

五、增量更新与存储 ​

5.1 每天增量拉取 ​

5.2 存储选型：CSV vs SQLite vs 数据库 ​

六、数据质量检查 ​

七、合法合规提醒 ​

八、下一步 ​

相关阅读 ​

02 · 数据获取实战

一、A 股数据：AKShare 与 Tushare

1.1 AKShare：免费、零门槛

1.2 Tushare Pro：更规范，需要积分

二、期货数据：主力合约 K 线

三、加密货币数据：Binance API

四、数据清洗：缺失值、停牌日与复权

4.1 缺失值

4.2 复权：前复权 / 后复权 / 不复权

五、增量更新与存储

5.1 每天增量拉取

5.2 存储选型：CSV vs SQLite vs 数据库

六、数据质量检查

七、合法合规提醒

八、下一步

相关阅读