大盘到底强不强？用 Python 做一个市场温度计

很多人判断市场强弱，习惯看指数：上证指数涨了，觉得行情不错；创业板跌了，觉得市场很差。

但指数有一个问题：它是少数权重股共同影响的结果。指数上涨，不代表大多数股票都在涨；指数下跌，也不代表没有结构性机会。

更好的办法是看市场广度，也就是回答几个更细的问题：

全市场有多少股票上涨？
有多少股票涨幅超过 3%？
有多少股票跌幅超过 3%？
多少股票站上 20 日均线？
成交额集中在少数股票，还是扩散到更多标的？

这篇文章用 AlphaFeed 写一个“市场温度计”：拉取全市场实时行情，再结合历史 K 线计算广度指标，输出一个简单但很有用的市场状态报告。

1. 为什么只看指数不够

假设某一天沪深 300 涨了 1%，你可能会觉得市场很强。

但真实情况可能有三种：

情况	市场含义
大多数股票上涨，指数也上涨	普涨，赚钱效应较好
少数权重股上涨，很多小票下跌	指数强，个股弱
指数小涨，但涨停家数很多	结构性行情活跃

如果只看指数，这三种情况会被混在一起。

所以做交易和研究时，我更喜欢看“市场内部结构”：上涨家数、下跌家数、强势股比例、弱势股比例、均线以上比例、成交额分布。

这些指标不神秘，Python 几十行代码就能算出来。关键是你需要一个能方便拿到全市场行情的数据源。

2. 用 AlphaFeed 获取全市场实时行情

安装：

pip install alphafeed pandas

初始化：

from alphafeed import AlphaFeed

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

也可以用环境变量：

export ALPHAFEED_API_KEY="your-api-key"

然后代码里直接：

from alphafeed import AlphaFeed

af = AlphaFeed()

获取全部 A 股实时行情：

import pandas as pd
from alphafeed import AlphaFeed

af = AlphaFeed()

quotes = af.quotes.get(universes=["CN_Stock"], to_dataframe=True)

print(len(quotes))
print(quotes[["symbol", "last_price", "prev_close", "volume", "amount", "ext.name", "ext.change_pct"]].head())

AlphaFeed 支持的标的池包括：

标的池	含义
`CN_Stock`	A 股
`CN_ETF`	ETF
`US_Stock`	美股
`HK_Stock`	港股

这意味着你可以用同样的方法做 A 股市场温度计、ETF 市场温度计、美股观察面板或港股观察面板。

3. 计算最基础的市场广度

先计算上涨、下跌、平盘数量：

def calc_basic_breadth(quotes):
    df = quotes.copy()
    change_pct = df["ext.change_pct"]

    total = len(df)
    up = int((change_pct > 0).sum())
    down = int((change_pct < 0).sum())
    flat = total - up - down

    return {
        "total": total,
        "up": up,
        "down": down,
        "flat": flat,
        "up_ratio": up / total if total else 0,
        "down_ratio": down / total if total else 0,
    }

打印结果：

breadth = calc_basic_breadth(quotes)

print(f"全市场标的数: {breadth['total']}")
print(f"上涨家数: {breadth['up']} ({breadth['up_ratio']:.1%})")
print(f"下跌家数: {breadth['down']} ({breadth['down_ratio']:.1%})")
print(f"平盘家数: {breadth['flat']}")

这个指标很简单，但已经比只看指数多了一层信息。

比如指数涨 0.5%，但上涨家数只有 25%，那说明行情可能只是权重股在撑指数。反过来，指数跌 0.3%，但上涨家数超过 60%，说明市场内部可能并不弱。

4. 计算强势股和弱势股比例

再定义几个阈值：

def calc_strength_distribution(quotes):
    df = quotes.copy()
    change_pct = df["ext.change_pct"]

    return {
        "rise_gt_3pct": int((change_pct >= 0.03).sum()),
        "rise_gt_5pct": int((change_pct >= 0.05).sum()),
        "fall_gt_3pct": int((change_pct <= -0.03).sum()),
        "fall_gt_5pct": int((change_pct <= -0.05).sum()),
    }

这些指标可以粗略判断赚钱效应：

指标	含义
涨幅超 3% 家数	主动进攻力量
涨幅超 5% 家数	强势股扩散程度
跌幅超 3% 家数	风险释放范围
跌幅超 5% 家数	恐慌程度

很多时候，大盘指数涨跌不大，但强势股数量会提前变化。市场温度计的价值就在这里：它不是预测明天，而是更客观地描述今天。

5. 找出成交额最高的股票

成交额代表市场注意力。我们可以看资金主要集中在哪里：

top_amount = quotes.sort_values("amount", ascending=False).head(20)

print(
    top_amount[[
        "symbol",
        "ext.name",
        "last_price",
        "ext.change_pct",
        "amount",
    ]]
)

如果每天成交额前 20 名都是少数权重股，说明市场可能偏指数行情；如果成交额扩散到很多中小市值股票，说明市场活跃度可能更高。

当然，AlphaFeed 这里只提供行情数据，不替你判断行业主题。你可以把成交额榜导出后，再结合自己的行业分类或观察列表做进一步分析。

6. 加入 20 日均线以上比例

实时行情能告诉我们今天涨跌，但判断趋势还需要历史 K 线。

我们可以抽取成交额较高的一批股票，计算有多少站上 20 日均线：

active_symbols = (
    quotes.sort_values("amount", ascending=False)
    .head(300)["symbol"]
    .tolist()
)

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

计算比例：

def calc_above_ma20_ratio(dfs):
    total = 0
    above = 0

    for symbol, df in dfs.items():
        df = df.sort_values("trade_date").copy()
        if len(df) < 20:
            continue

        ma20 = df["close"].rolling(20).mean().iloc[-1]
        close = df["close"].iloc[-1]

        if pd.isna(ma20):
            continue

        total += 1
        if close > ma20:
            above += 1

    return above / total if total else 0

使用：

above_ma20_ratio = calc_above_ma20_ratio(dfs)
print(f"活跃股票中站上20日线比例: {above_ma20_ratio:.1%}")

这个指标很适合做市场温度判断：

20 日线以上比例	粗略状态
70% 以上	市场较热
50% - 70%	偏强或震荡偏强
30% - 50%	震荡偏弱
30% 以下	市场较冷

阈值只是经验，不要机械使用。不同市场、不同时期都需要校准。

7. 输出一份市场温度报告

把上面的部分合起来：

import pandas as pd
from datetime import datetime
from alphafeed import AlphaFeed


def calc_basic_breadth(quotes):
    change_pct = quotes["ext.change_pct"]
    total = len(quotes)
    up = int((change_pct > 0).sum())
    down = int((change_pct < 0).sum())
    flat = total - up - down
    return total, up, down, flat


def calc_strength_distribution(quotes):
    change_pct = quotes["ext.change_pct"]
    return {
        "rise_gt_3pct": int((change_pct >= 0.03).sum()),
        "rise_gt_5pct": int((change_pct >= 0.05).sum()),
        "fall_gt_3pct": int((change_pct <= -0.03).sum()),
        "fall_gt_5pct": int((change_pct <= -0.05).sum()),
    }


def calc_above_ma20_ratio(dfs):
    total = 0
    above = 0
    for _, df in dfs.items():
        df = df.sort_values("trade_date").copy()
        if len(df) < 20:
            continue
        ma20 = df["close"].rolling(20).mean().iloc[-1]
        close = df["close"].iloc[-1]
        if pd.isna(ma20):
            continue
        total += 1
        above += int(close > ma20)
    return above / total if total else 0


def market_temperature_label(up_ratio, above_ma20_ratio, weak_count):
    if up_ratio >= 0.65 and above_ma20_ratio >= 0.6:
        return "偏热"
    if up_ratio >= 0.5 and above_ma20_ratio >= 0.45:
        return "偏强"
    if up_ratio <= 0.35 or weak_count > 500:
        return "偏冷"
    return "震荡"


def build_market_report(universe="CN_Stock"):
    af = AlphaFeed()

    quotes = af.quotes.get(universes=[universe], to_dataframe=True)
    quotes = quotes.dropna(subset=["ext.change_pct", "amount"])

    total, up, down, flat = calc_basic_breadth(quotes)
    dist = calc_strength_distribution(quotes)

    active_symbols = (
        quotes.sort_values("amount", ascending=False)
        .head(300)["symbol"]
        .tolist()
    )
    dfs = af.klines.batch(
        active_symbols,
        period="1d",
        count=60,
        adjust="forward",
        to_dataframe=True,
        show_progress=True,
    )
    above_ma20_ratio = calc_above_ma20_ratio(dfs)

    up_ratio = up / total if total else 0
    label = market_temperature_label(
        up_ratio,
        above_ma20_ratio,
        dist["fall_gt_3pct"],
    )

    top_amount = quotes.sort_values("amount", ascending=False).head(10)

    lines = []
    lines.append(f"# 市场温度计 {datetime.now():%Y-%m-%d %H:%M}")
    lines.append("")
    lines.append(f"- 标的池：{universe}")
    lines.append(f"- 市场状态：{label}")
    lines.append(f"- 上涨家数：{up} / {total} ({up_ratio:.1%})")
    lines.append(f"- 下跌家数：{down} / {total} ({down / total:.1%})")
    lines.append(f"- 平盘家数：{flat}")
    lines.append(f"- 涨幅超过 3%：{dist['rise_gt_3pct']}")
    lines.append(f"- 涨幅超过 5%：{dist['rise_gt_5pct']}")
    lines.append(f"- 跌幅超过 3%：{dist['fall_gt_3pct']}")
    lines.append(f"- 跌幅超过 5%：{dist['fall_gt_5pct']}")
    lines.append(f"- 活跃股票站上 20 日线比例：{above_ma20_ratio:.1%}")
    lines.append("")
    lines.append("## 成交额 Top 10")
    lines.append("")
    lines.append(top_amount[["symbol", "ext.name", "last_price", "ext.change_pct", "amount"]].to_markdown(index=False))
    lines.append("")
    lines.append("数据来源：AlphaFeed (https://alphafeed.org/)")

    return "\n".join(lines)


if __name__ == "__main__":
    report = build_market_report("CN_Stock")
    print(report)

    with open("market-temperature.md", "w", encoding="utf-8") as f:
        f.write(report)

如果需要 to_markdown，安装：

pip install tabulate

8. 这个报告怎么读

你可以每天收盘后生成一次报告，然后观察几个核心变化：

指标	观察重点
上涨家数比例	市场赚钱效应是否扩散
跌幅超过 3% 家数	风险释放是否扩大
活跃股站上 20 日线比例	趋势是否健康
成交额 Top 10	资金是否集中在少数权重股
市场状态标签	给自己一个快速摘要

不要把它当成买卖信号。它更像天气预报里的温度和湿度：告诉你环境如何，但不会替你决定今天穿哪件衣服。

交易决策仍然要结合你的策略、仓位、风险承受能力和交易计划。

9. 扩展到 ETF、美股、港股

如果你想看 ETF 市场：

report = build_market_report("CN_ETF")

如果你想看美股：

report = build_market_report("US_Stock")

如果你想看港股：

report = build_market_report("HK_Stock")

跨市场统一接口的好处在这里会非常明显。你的市场温度计逻辑不用重写，只要换一个标的池 ID，就可以观察不同市场。

当然，不同市场的交易制度、涨跌幅分布、成交结构都不一样，所以阈值不要照搬。比如 A 股和美股的单日波动结构不同，涨幅超过 3% 在不同市场里的含义也不同。

10. 可以继续做成一个 Dashboard

这篇文章输出的是 Markdown 报告。如果你想更进一步，可以做成网页看板：

用 Streamlit 显示市场温度；
用 Plotly 画上涨/下跌比例柱状图；
每 5 分钟刷新一次实时行情；
保存每日市场温度到 CSV；
画出市场温度历史曲线；
给极端冷/热状态加提醒。

你甚至可以让 AI 编程工具帮你改：

请基于这个 AlphaFeed 市场温度计脚本，帮我改成 Streamlit Dashboard：
1. 页面顶部显示市场状态标签；
2. 显示上涨/下跌/平盘数量；
3. 显示涨幅超过 3%、跌幅超过 3% 的数量；
4. 显示成交额 Top 20 表格；
5. 支持在侧边栏选择 CN_Stock、CN_ETF、US_Stock、HK_Stock。

这种需求非常适合 AlphaFeed + AI 编程：AlphaFeed 提供结构化数据，AI 帮你快速搭界面，你把注意力放在指标设计和结果解释上。

11. 为什么这是一个适合普通人的量化工具

很多人一听量化，就觉得一定要机器学习、深度学习、高频交易。其实不是。

对大多数个人投资者和研究者来说，更实用的量化工具往往是：

自动整理市场信息；
用固定规则减少主观情绪；
长期记录市场状态；
用数据验证自己的感觉；
在风险变大时及时降温。

市场温度计就是这样的工具。它不复杂，但每天都能用。

当你觉得“今天市场好像很强”时，可以看上涨家数比例；当你觉得“指数没跌多少，但账户很难受”时，可以看跌幅超过 3% 的家数；当你觉得“是不是该加仓”时，可以先看活跃股票站上 20 日线的比例。

它不能替你赚钱，但能帮你少一点拍脑袋。

结语

判断市场强弱，不应该只盯着指数涨跌。指数是结果，市场广度才更接近交易者真实体感。

用 AlphaFeed，你可以用几行 Python 拿到全市场实时行情，再结合历史 K 线计算自己的市场温度指标。这个过程没有玄学，也不需要复杂模型，核心就是把“感觉”变成“可重复观察的数据”。

如果你正在搭建自己的量化工作流，可以从这个市场温度计开始：每天生成一份报告，持续记录，慢慢你会对市场状态有更稳定的判断。

参考文献：

AlphaFeed 官网：https://alphafeed.org/
AlphaFeed 文档：https://docs.alphafeed.org/

大盘到底强不强？用 Python 做一个市场温度计

大盘到底强不强？用 Python 做一个市场温度计

1. 为什么只看指数不够

2. 用 AlphaFeed 获取全市场实时行情

3. 计算最基础的市场广度

4. 计算强势股和弱势股比例

5. 找出成交额最高的股票

6. 加入 20 日均线以上比例

7. 输出一份市场温度报告

8. 这个报告怎么读

9. 扩展到 ETF、美股、港股

10. 可以继续做成一个 Dashboard

11. 为什么这是一个适合普通人的量化工具

结语

评论