普通人如何 从 0 到 1 跑一个可复现的量化策略

很多人第一次学量化，会卡在一个非常朴素的问题上：数据到底从哪里来？

教程里经常一上来就讲均线、因子、回测框架、夏普比率，但真正开始写代码时，第一步往往是最烦的：股票代码格式不统一，日 K 线字段不稳定，复权方式说不清，实时行情和历史行情又来自不同接口。结果策略还没开始，时间先耗在了“修数据”上。

AlphaFeed 想解决的正是这件事：把 A 股、美股、港股等市场的行情数据，用一个统一的 Python SDK 暴露出来。你只需要会一点 Python，就可以把数据拉成 pandas DataFrame，然后开始写自己的策略逻辑。

这篇文章不讲玄学，也不讲“稳赚模型”。我们只做一件实际的事：用 AlphaFeed 获取股票 K 线，写一个最基础的均线趋势策略，并做一个轻量回测。代码可以作为你后续研究更复杂策略的起点。

1. 先准备数据源

安装 SDK：

pip install alphafeed pandas matplotlib

初始化：

from alphafeed import AlphaFeed

af = AlphaFeed(api_key="your-api-key")

也可以把 key 放到环境变量里：

export ALPHAFEED_API_KEY="your-api-key"

然后代码里直接写：

from alphafeed import AlphaFeed

af = AlphaFeed()

AlphaFeed 的标的代码格式比较直接：

市场	示例
A 股上交所	600519.SH
A 股深交所	000001.SZ
美股	AAPL.US
港股	00700.HK

这对写策略很重要。因为策略代码最怕“同一只股票在不同数据源里长得不一样”：有的用 sh600519，有的用 600519，有的用 600519.SH。格式不统一，后面批量回测、组合管理、日志记录都会变麻烦。

2. 获取一只股票的日 K 线

先拉贵州茅台最近 300 根日 K：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

df = af.klines.get(
    "600519.SH",
    period="1d",
    count=300,
    adjust="forward",
    to_dataframe=True,
)

df = df.sort_values("trade_date").reset_index(drop=True)
print(df.tail())

返回的 DataFrame 里常用字段包括：

字段	含义
trade_date	交易日期
open	开盘价
high	最高价
low	最低价
close	收盘价
volume	成交量
amount	成交额

这里我使用了 adjust="forward"，也就是前复权。做中长期股票策略时，复权方式会影响收益率计算和均线位置。AlphaFeed 支持多种复权方式：

df = af.klines.get("600519.SH", adjust="forward", to_dataframe=True)
df = af.klines.get("600519.SH", adjust="backward", to_dataframe=True)
df = af.klines.get("600519.SH", adjust="forward_additive", to_dataframe=True)
df = af.klines.get("600519.SH", adjust="backward_additive", to_dataframe=True)
df = af.klines.get("600519.SH", adjust="none", to_dataframe=True)

新手阶段建议先固定一种复权方式，不要一会儿用前复权，一会儿用不复权。否则你看到的策略结果可能只是数据处理方式变化带来的错觉。

3. 写一个最基础的双均线策略

策略规则：

1. 计算 20 日均线和 60 日均线。
2. 当 20 日均线高于 60 日均线时，认为趋势偏强，持有。
3. 当 20 日均线低于 60 日均线时，空仓。
4. 为避免未来函数，今天产生的信号，明天才用于计算收益。

代码如下：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

symbol = "600519.SH"

df = af.klines.get(
    symbol,
    period="1d",
    count=800,
    adjust="forward",
    to_dataframe=True,
)

df = df.sort_values("trade_date").reset_index(drop=True)

df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()

# 原始信号：短均线在长均线上方则持有
df["signal"] = (df["ma20"] > df["ma60"]).astype(int)

# 关键：信号后移一天，避免用今天收盘价决定今天是否持仓
df["position"] = df["signal"].shift(1).fillna(0)

df["ret"] = df["close"].pct_change().fillna(0)
df["strategy_ret"] = df["position"] * df["ret"]

df["equity"] = (1 + df["strategy_ret"]).cumprod()
df["buy_and_hold"] = (1 + df["ret"]).cumprod()

print(df[["trade_date", "close", "ma20", "ma60", "position", "equity"]].tail())
print("策略累计收益:", df["equity"].iloc[-1] - 1)
print("买入持有收益:", df["buy_and_hold"].iloc[-1] - 1)

这个策略很简单，但它已经包含了策略研究最核心的几个动作：

动作	为什么重要
获取稳定数据	策略研究必须先保证输入可靠
明确复权方式	避免分红送转造成价格序列断裂
构造交易信号	把想法变成可执行规则
信号延迟一日	避免未来函数
计算净值曲线	用结果检验策略，而不是用感觉判断

4. 加上最大回撤和年化指标

只看累计收益不够。一个策略赚了 50%，但中间最大回撤 45%，大多数人是拿不住的。

我们继续补几个指标：

import numpy as np

def annual_return(equity: pd.Series, periods_per_year: int = 252) -> float:
    total_return = equity.iloc[-1] / equity.iloc[0] - 1
    years = len(equity) / periods_per_year
    return (1 + total_return) ** (1 / years) - 1

def max_drawdown(equity: pd.Series) -> float:
    peak = equity.cummax()
    drawdown = equity / peak - 1
    return drawdown.min()

def sharpe_ratio(returns: pd.Series, periods_per_year: int = 252) -> float:
    if returns.std() == 0:
        return 0
    return returns.mean() / returns.std() * np.sqrt(periods_per_year)

print("年化收益:", annual_return(df["equity"]))
print("最大回撤:", max_drawdown(df["equity"]))
print("夏普比率:", sharpe_ratio(df["strategy_ret"]))

这一步开始，你就不再只是“看图炒股”，而是在建立一个可复现、可比较的研究流程。

5. 扩展到多只股票

单只股票的策略表现，很容易被个股特性影响。下一步应该做批量测试。

AlphaFeed 支持批量获取 K 线：

from alphafeed import AlphaFeed

af = AlphaFeed()

symbols = ["600519.SH", "000001.SZ", "601318.SH", "AAPL.US", "00700.HK"]

dfs = af.klines.batch(
    symbols,
    period="1d",
    count=800,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

print(dfs.keys())

可以把刚才的策略封装成函数：

def run_ma_strategy(df: pd.DataFrame, short_window=20, long_window=60) -> dict:
    df = df.sort_values("trade_date").reset_index(drop=True).copy()
    df["ma_short"] = df["close"].rolling(short_window).mean()
    df["ma_long"] = df["close"].rolling(long_window).mean()
    df["signal"] = (df["ma_short"] > df["ma_long"]).astype(int)
    df["position"] = df["signal"].shift(1).fillna(0)
    df["ret"] = df["close"].pct_change().fillna(0)
    df["strategy_ret"] = df["position"] * df["ret"]
    df["equity"] = (1 + df["strategy_ret"]).cumprod()

    return {
        "total_return": df["equity"].iloc[-1] - 1,
        "annual_return": annual_return(df["equity"]),
        "max_drawdown": max_drawdown(df["equity"]),
        "sharpe": sharpe_ratio(df["strategy_ret"]),
    }

results = []
for symbol, one_df in dfs.items():
    metrics = run_ma_strategy(one_df)
    metrics["symbol"] = symbol
    results.append(metrics)

result_df = pd.DataFrame(results).sort_values("sharpe", ascending=False)
print(result_df)

这样你就可以回答更具体的问题：

1. 这个策略只对某只股票有效，还是在一组股票上都有一定稳定性？
2. A 股、美股、港股里的表现差异是什么？
3. 均线参数换成 10/30、20/120 后，结果是否仍然合理？
4. 策略收益来自趋势跟随，还是来自某段行情的偶然上涨？

6. 用实时行情做一个盘中观察列表

历史回测之外，实盘观察也需要实时行情。比如我们想看几只股票当前涨跌幅：

from alphafeed import AlphaFeed

af = AlphaFeed()

watchlist = ["600519.SH", "000001.SZ", "601318.SH", "AAPL.US", "00700.HK"]

quotes = af.quotes.get(symbols=watchlist, to_dataframe=True)

quotes["change_pct_calc"] = (
    quotes["last_price"] - quotes["prev_close"]
) / quotes["prev_close"]

cols = ["symbol", "last_price", "prev_close", "change_pct_calc", "volume", "trade_time"]
print(quotes[cols].sort_values("change_pct_calc", ascending=False))

如果你订阅了相应权限，也可以直接获取某个市场的全量行情池：

all_cn = af.quotes.get(universes="CN_Stock", to_dataframe=True)
print("A 股数量:", len(all_cn))
print(all_cn[["symbol", "last_price", "volume", "trade_time"]].head())

这时，你可以进一步做盘中筛选：

active = all_cn[
    (all_cn["volume"] > all_cn["volume"].quantile(0.8)) &
    (all_cn["last_price"] > all_cn["prev_close"])
]

print(active[["symbol", "last_price", "prev_close", "volume"]].head(20))

7. 新手最容易犯的 5 个错误

第一，直接用当天信号计算当天收益。

如果你用收盘价计算均线，又用同一天的收盘价计算收益，本质上就是提前知道了收盘结果。解决方法是把 signal 后移一天。

第二，只在一只股票上调参数。

你把 20/60 调到 17/43，可能只是刚好适配了某只股票的历史走势。真正要看的是一组股票、多段行情、不同市场里的稳定性。

第三，忽略复权。

股票分红、送转、拆股会改变价格序列。做历史回测时，复权方式必须固定并写进研究记录。

第四，只看收益，不看回撤。

策略的第一任务不是“历史上最高赚多少”，而是“在可承受波动下是否有稳定优势”。

第五，把数据清洗当成策略能力。

很多人在数据源之间来回搬运、改字段、补异常值，最后误以为自己在做策略研究。真正的研究时间应该花在假设、验证、归因和风险控制上，而不是每天修接口。

8. 一个更实用的学习路线

如果你是普通人，想从零开始做量化，不建议先啃厚书，也不建议一上来写高频系统。可以按这个顺序来：

1. 用 AlphaFeed 拉一只股票的 K 线，画出收盘价。
2. 写一个均线策略，理解信号、持仓、收益、回撤。
3. 扩展到 20 只股票，观察策略是否稳定。
4. 加入交易成本和滑点。
5. 把策略拆成数据层、信号层、回测层。
6. 再学习因子、组合、风险预算和参数检验。

这条路线的关键是：每一步都有代码、有数据、有结果。你不需要先成为金融工程专家，才开始做第一轮实验。

结语

量化不是神秘学，它更像一套可复现的研究流程。

AlphaFeed 的价值在于，把最容易消耗耐心的数据接入问题，压缩成几行 Python 代码。你可以更快地进入真正重要的部分：提出假设、写出规则、跑出结果、检查风险，然后继续迭代。

从这个角度看，第一篇量化代码不需要复杂。只要它能稳定获取数据、能避免未来函数、能输出收益和回撤，它就已经是一个合格的起点。

相关链接：

• AlphaFeed 官网：AlphaFeed
• Python SDK 快速开始：AlphaFeed Quickstart
• Python SDK GitHub/README：可参考项目中的 alphafeed-python-sdk/README.md