金猪博客 金猪博客

把市场研究做成程序：用 Python 搭一条可复现的美股行业轮动流水线

Chen Xi
Chen Xi
  2026-08-08 14:30:00   2026-08-08 14:30:00  

写完一篇市场报告之后，我通常会有一点不踏实。

当时的判断可能有数据、有图表，也有一套听起来完整的逻辑，但过两周再回头看，很难回答一个简单的问题：如果把今天的分析重新跑一遍，结论还会不会一样？

问题往往不在公式，而在过程。数据从哪里来、什么时候拿到的、有没有被修订、不同频率怎么对齐、缺失值是怎么处理的，这些细节如果没有留下来，文章就只能算一次性的观察，不能算研究。

所以这篇文章不做“下个月哪个板块会涨”的预测。我想把一个更朴素的流程搭出来：每个月把宏观数据和行业价格放进同一张表，计算一套透明的动量分数，画出热力图，然后把当时的判断和数据版本一起保存下来。

image

先把问题缩小到能验证的范围

“研究美股行业轮动”听起来很大，真正落到代码里，我只保留三个问题：

  1. 最近 3 个月和 12 个月，哪些行业的相对表现更强？
  2. 这个排名是在上涨、横盘还是下跌的市场里出现的？
  3. 如果换一个数据截止日，排名会不会完全变样？

这三个问题有一个共同点：它们都能被保存成输入、函数和输出。至于“下个月应该买什么”，不在这条流水线的职责范围内。

这里的行业可以用行业指数，也可以用代表性 ETF 作为代理。关键不是选一个看起来最漂亮的标的，而是从头到尾保持口径一致。示例热力图里的数值是演示数据，不代表当前市场表现，也不构成投资建议。

数据源要朴素，而且要记得当时的版本

我会把数据源分成三层，而不是把所有字段都塞进一个行情接口：

层次 数据 用途
市场价格 行业指数或 ETF 的日收盘价 计算月度收益、动量和回撤
宏观环境 FRED 的 10 年期国债收益率、CPI、失业率 描述利率和经济背景
实体经济 BEA 的 GDP、PCE、企业利润 检查价格走势背后的需求和盈利

FRED API 提供 REST 接口，可以返回 JSON 或 XML；如果需要严格复盘，还可以使用 ALFRED 的实时版本，避免把后来修订过的数据误当成当时已经知道的数字。 BEA 的开发者接口 则适合按程序读取 GDP、个人消费和企业利润等统计数据。

API 密钥不要写进文章里的代码，更不要提交到仓库。我的做法是把它放到环境变量里：

1
$env:FRED_API_KEY = "你的密钥"

价格数据则单独保存为 CSV。这个文件可以来自你有权限使用的数据服务，也可以来自自己整理的指数历史数据。不要为了省几分钟去抓取一个经常改版的网页，网页展示层一变，整个研究就会跟着坏掉。

先定一个不容易出错的价格表

只要字段稳定，后面的计算其实很简单。原始文件至少保留这三列：

1
2
3
4
date,symbol,close
2026-01-30,XLK,241.20
2026-01-30,XLE,91.48
2026-01-30,XLI,144.05

date 是交易日期，symbol 是行业代码，close 是经过明确说明的收盘价。是否使用复权价格，要在数据字典里写清楚；同一份研究里不能一部分用普通收盘价，另一部分用复权价。

读取时先做几项很无聊、但很有用的检查：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from pathlib import Path

import pandas as pd


REQUIRED_COLUMNS = {"date", "symbol", "close"}


def load_prices(path: str | Path) -> pd.DataFrame:
    frame = pd.read_csv(path)
    missing = REQUIRED_COLUMNS - set(frame.columns)
    if missing:
        raise ValueError(f"缺少字段: {sorted(missing)}")

    frame["date"] = pd.to_datetime(frame["date"], errors="coerce")
    frame["close"] = pd.to_numeric(frame["close"], errors="coerce")
    frame["symbol"] = frame["symbol"].astype("string").str.strip().str.upper()
    frame = frame.dropna(subset=["date", "symbol", "close"])

    if (frame["close"] <= 0).any():
        raise ValueError("收盘价必须大于 0")
    if frame.duplicated(["date", "symbol"]).any():
        raise ValueError("同一个交易日存在重复代码")

    return frame.sort_values(["symbol", "date"]).reset_index(drop=True)

这里没有急着填充缺失日期。交易日不是自然日，周末和节假日不应该被伪造出来；真正需要对齐到月度时，再按每个行业最后一个有效交易日取值。

从日线变成月度收益

行业轮动不需要每天换一次观点。先把日线压成月末价格，再计算收益，结果更容易解释，也能减少短期噪声。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def monthly_returns(prices: pd.DataFrame) -> pd.DataFrame:
    data = prices.copy()
    data["month"] = data["date"].dt.to_period("M")

    # 每个行业每个月只保留最后一个有效交易日
    month_end = (
        data.sort_values("date")
        .groupby(["symbol", "month"], as_index=False)
        .tail(1)
    )
    close = month_end.pivot(index="month", columns="symbol", values="close")
    close = close.sort_index()
    return close.pct_change(fill_method=None)


prices = load_prices("data/sector_prices.csv")
returns = monthly_returns(prices)
three_month = returns.rolling(3, min_periods=3).sum()
twelve_month = returns.rolling(12, min_periods=12).sum()

这里使用的是简单收益率的滚动累加，足够作为第一版观察指标。如果你把它用于正式回测，需要改成复合收益：

1
2
3
4
twelve_month = (1 + returns).rolling(12, min_periods=12).apply(
    lambda window: window.prod() - 1,
    raw=True,
)

两种写法的差异在极端行情里会变大。把公式写在代码里并留下版本号，比在文章里只放一个最终数字更重要。

把宏观数据接进来，但不让它偷看未来

FRED 的接口返回的原始观测值里，缺失值通常会以 . 表示。读取之后先转成数值，再按月聚合：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import os

import requests


def fetch_fred(series_id: str, start: str) -> pd.Series:
    params = {
        "series_id": series_id,
        "api_key": os.environ["FRED_API_KEY"],
        "file_type": "json",
        "observation_start": start,
    }
    response = requests.get(
        "https://api.stlouisfed.org/fred/series/observations",
        params=params,
        timeout=20,
    )
    response.raise_for_status()
    observations = pd.DataFrame(response.json()["observations"])
    observations["date"] = pd.to_datetime(observations["date"])
    observations["value"] = pd.to_numeric(observations["value"], errors="coerce")
    return observations.set_index("date")["value"].dropna().sort_index()


ten_year = fetch_fred("DGS10", "2018-01-01")
unemployment = fetch_fred("UNRATE", "2018-01-01")

# 与价格表使用相同的 PeriodIndex，避免用自然月的最后一天硬拼
macro = pd.DataFrame({
    "ten_year": ten_year,
    "unemployment": unemployment,
})
macro["month"] = macro.index.to_period("M")
macro = macro.groupby("month").last()

真正容易出问题的是发布日期。比如一份 6 月的就业数据可能在 7 月才公布；如果用后来下载到的 6 月数值去解释 6 月底的投资决策，就已经产生了未来函数。正式研究时，我会额外保存 observation_date、release_date 和 vintage_date 三列；没有这些字段，最多只能做事后描述，不能声称自己做了实时回测。

宏观变量也不应该直接变成“买入信号”。它更适合用来标记环境，例如把 10 年期利率变化画在行业热力图上方，提醒自己：一次排名变化可能来自折现率，而不是行业盈利突然改善。

用相对排名代替拍脑袋的强弱判断

我会把 3 个月和 12 个月动量转成横截面 z-score，再做一个透明的加权分数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def cross_sectional_zscore(frame: pd.DataFrame) -> pd.DataFrame:
    mean = frame.mean(axis=1)
    std = frame.std(axis=1, ddof=0).mask(lambda values: values == 0)
    return frame.sub(mean, axis=0).div(std, axis=0)


def build_score(returns: pd.DataFrame) -> pd.DataFrame:
    fast = returns.rolling(3, min_periods=3).sum()
    slow = (1 + returns).rolling(12, min_periods=12).apply(
        lambda window: window.prod() - 1,
        raw=True,
    )
    return 0.6 * cross_sectional_zscore(fast) + 0.4 * cross_sectional_zscore(slow)


score = build_score(returns)
latest = score.iloc[-1].sort_values(ascending=False).dropna()
print(latest.to_string(float_format=lambda value: f"{value: .2f}"))

这里的 0.6 和 0.4 只是可解释的起点，不是经过神奇优化得到的参数。参数越多，越容易把历史噪声调成一套看似精准的规则。第一版研究宁可简单一点，先确认每个月的输入、输出和异常都能复核，再讨论是否需要增加波动率、估值或盈利修正。

image

热力图要服务于判断，而不是装饰

一张热力图最有用的地方，不是颜色够不够鲜艳，而是能不能一眼看出排名有没有持续性。我的习惯是：

  • 颜色表示月度收益，数字保留一位小数；
  • 行业顺序按最后一个月的分数排序，而不是按字母排序；
  • 图表标题写明数据截止日，避免截图在半年后失去上下文；
  • 对演示数据明确标注“illustrative”，不让读者把示例误认为实时行情。

如果要生成自己的图，可以用 seaborn.heatmap，但不要把绘图过程和数据下载绑死在一个脚本里。比较舒服的目录结构是：

1
2
3
4
5
6
research/
├─ raw/          # 原始响应，不覆盖
├─ cleaned/      # 清洗后的 parquet/csv
├─ figures/      # 热力图和检查图
├─ reports/      # 每次运行的摘要
└─ run_pipeline.py

每次运行至少保存 as_of、数据源 URL、代码版本和行数统计。这样下次图形变了，先能回答“数据变了还是代码变了”，而不是重新猜一遍。

我会专门记录的四类失败

1. 把公布日期当成观测日期

这是最隐蔽的错误。宏观数据的月份和它真正公布的日期不是一回事。解决办法是保留发布日，或者直接使用 ALFRED 的 vintage 数据。

2. 用后来的成分股回看过去

如果今天的行业 ETF 成分被拿去解释十年前的行业表现，就引入了幸存者偏差。指数历史数据和成分股历史数据要分开，不能因为代码能跑就当作口径正确。

3. 用普通收盘价拼出不存在的收益

分红、拆分和基金分配都会影响价格序列。研究总收益就使用明确的复权口径，研究价格趋势就不要把结果写成“投资者实际赚了多少”。

4. 只看最后一张图

最后一张图通常最漂亮，也最容易让人忘记中间发生过什么。保留每次运行的 CSV、日志和异常记录，才能看到某个行业是持续强，还是只被一个月的异常值推上去。

把一次分析变成每月都能重跑的工具

这条流水线最值得留下来的，不是某个月的第一名，而是一个可重复的习惯：

  1. 先冻结数据截止日，再下载数据；
  2. 原始文件只追加，不覆盖；
  3. 清洗、计算和绘图分成不同函数；
  4. 输出排名的同时输出缺失值、重复值和数据范围；
  5. 把结论写在图表旁边，也把失效条件写出来。

当一个月后的新数据进来时，我只需要重新运行一次脚本，就能看到旧判断哪些仍然成立，哪些只是当时的噪声。这种节奏比每天追着行情刷新图表更接近研究，也更适合个人博客持续积累内容。

最后的话

程序不会自动把市场变简单。它能做的，是把“我感觉这个板块变强了”拆成一组可以检查的步骤：哪一天的数据、哪一种收益、哪一个时间窗口、哪一个参数，最后得到哪一张图。

如果这套流程最后只能告诉我“过去三个月能源相对强”，它依然有价值。至少我知道这句话是怎么来的，也知道什么新数据出现时应该推翻它。

这篇文章讨论的是数据整理和研究方法，不构成任何投资、交易或税务建议。市场数据会修订，历史排名也不代表未来表现；使用真实资金前，请单独核对数据授权、交易成本、税务和风险承受能力。

参考资料

把市场研究做成程序：用 Python 搭一条可复现的美股行业轮动流水线
2026/08/08/python-market-pipeline/
Author
Chen Xi
Published
2026-08-08 14:30
License
  1. 先把问题缩小到能验证的范围
  2. 数据源要朴素，而且要记得当时的版本
  3. 先定一个不容易出错的价格表
  4. 从日线变成月度收益
  5. 把宏观数据接进来，但不让它偷看未来
  6. 用相对排名代替拍脑袋的强弱判断
  7. 热力图要服务于判断，而不是装饰
  8. 我会专门记录的四类失败
    1. 1. 把公布日期当成观测日期
    2. 2. 用后来的成分股回看过去
    3. 3. 用普通收盘价拼出不存在的收益
    4. 4. 只看最后一张图
  9. 把一次分析变成每月都能重跑的工具
  10. 最后的话
  11. 参考资料
  1. 先把问题缩小到能验证的范围
  2. 数据源要朴素，而且要记得当时的版本
  3. 先定一个不容易出错的价格表
  4. 从日线变成月度收益
  5. 把宏观数据接进来，但不让它偷看未来
  6. 用相对排名代替拍脑袋的强弱判断
  7. 热力图要服务于判断，而不是装饰
  8. 我会专门记录的四类失败
    1. 1. 把公布日期当成观测日期
    2. 2. 用后来的成分股回看过去
    3. 3. 用普通收盘价拼出不存在的收益
    4. 4. 只看最后一张图
  9. 把一次分析变成每月都能重跑的工具
  10. 最后的话
  11. 参考资料