很多投资者在看直播时最容易记住一句话结论但真正能留在工具箱里的是把“市场温度”变成一套可重复计算、可批量回测、可自动更新的程序化指标。当前 A 股和港股的讨论里“泡沫”“热度”“高低切换”更多来自情绪表达而程序化分析要做到的是用数据给“温度”定标尺。这次我们来看的是一套基于 Python 的 A 股与港股市场温度分析框架。它不依赖任何昂贵行情终端也没有 GPU 和显存门槛只需要一台普通电脑从公开行情接口拉数据再用 pandas 做聚合计算就能产出市场热度评分、趋势序列、历史分位图表和 JSON 接口。整个过程可以全部自动化跑成每日定时任务也可以做成 HTTP 服务供其他程序调用。这篇文章会直接给出核心指标的拆解、数据获取方式、温度计算代码、批量回测思路、接口化改造和常见问题排查。读完你不需要掌握复杂的金融模型只要会一点 Python 和 pandas跟着步骤就能把整套分析流程跑起来并且能用自己的数据源替换示例接口按实际交易品种调整参数。文章结尾还会给出合规使用边界避免把数据分析工具变成盲目跟风的依据。1. 市场温度分析系统核心能力速览能力项说明市场覆盖A 股、港股可按需要扩充其他市场数据来源公开行情接口需要按实际可用数据源配置核心指标涨跌家数比例、成交额分位数、估值百分位、波动率水平运行环境Python pandas NumPy Matplotlib无需 GPU输出形式评分表 CSV / Excel / Markdown趋势图 PNGJSON 接口批量能力支持按天、周、月批量扫描历史区间输出温度序列接口能力可基于 FastAPI 暴露 HTTP JSON API供其他程序调用定时任务支持 Windows 计划任务或 Linux Cron 自动执行硬件要求普通 PC8GB 内存即可顺畅运行核心优势指标可解释、代码可修改、结果可回测这套系统的核心思想很简单温度不是单一看点而是多个维度合成的结果。单独看指数涨跌容易被权重股带动单独看成交额又忽略估值位置所以要把市场广度、成交活跃度、估值水平、波动状态放在一起评分再判断当前处于偏冷、适中还是偏热区间。如果从输入材料来看市场直播里提到的“泡沫中洗澡”更多是一种主观类比本文不讨论对错而是把“泡沫”翻译成可检验的量化条件。例如估值分位处于历史高位、成交额分位同步抬升、上涨家数占比超过某阈值三者同时出现才在程序上标记为“温度过热”。这样的框架优势在于下一次市场再被讨论“温度高还是低”时你可以直接用同一套代码复核而不需要猜。2. 适用场景与使用边界2.1 适合谁使用这套框架适合四类人群。第一类是量化研究初学者想用 Python 对 A 股、港股做系统性的行情观察第二类是个人投资者每天复盘时不想只看指数涨跌幅还想知道市场内部是普涨还是分化第三类是内容创作者或投研人员需要批量生成市场温度图表和统计口径给报告或视频提供数据支撑第四类是程序开发者想把行情分析能力接入自己的工具、机器人或内部系统。2.2 能解决的问题它能解决三个实际痛点。第一个痛点是口径不统一不同人讨论“温度”时参考的指标完全不同而程序化框架可以固定一批指标保证每次分析口径一致。第二个痛点是重复劳动每天手动收集涨跌家数、成交额、估值数据很费时间脚本加定时任务可以自动汇总。第三个痛点是缺少历史参照当前市场到底是热是冷必须对比历史分位数才有意义批量回测可以快速算出一段区间内的温度序列。2.3 不适合什么场景这套框架不适合作为短期买卖信号不适合预测具体点位也不适合替代基本面研究。市场温度是慢变量更适合做仓位参考和风险提示而不是日内交易触发器。温度偏高只表示历史统计意义上的过热不代表马上回调温度偏低也不代表一定反弹。所有输出都只是统计描述不是投资建议。2.4 版权、隐私与安全边界涉及行情数据时要确认所使用数据源的授权协议个人研究通常允许使用公开接口但商用场景必须购买合规数据服务。不得把内部行情数据、客户数据或未公开信息放入自动化处理流程中传递。任何基于该框架生成的结论在公开发布前都需要人工复核不能把程序输出直接当作事实发布。市场有风险本文全部内容仅作为技术方法交流不构成任何投资建议或操作指令。3. 环境准备与数据源配置3.1 操作系统与 Python 环境推荐使用 Windows 10/11、Ubuntu 20.04 或 macOS 12 以上版本。Python 建议使用 3.9 到 3.11 之间避免过新版本导致部分第三方库兼容问题。不需要 GPU也不需要安装 CUDA。首先创建独立虚拟环境避免污染系统 Pythonpython -m venv markettemp_venv source markettemp_venv/bin/activate # Windows 下使用 markettemp_venv\Scripts\activate然后再安装核心依赖。需要说明的是以下命令属于通用安装流程具体包版本以当前 PyPI 实际可用版本为准pip install pandas numpy matplotlib akshare fastapi uvicorn schedule openpyxl其中akshare是常用的中国金融市场公开数据接口库但接口命名和字段可能随版本变化。如果某个接口不存在或返回字段发生变化需要查阅当前版本的官方文档。也可以替换为 tushare、baostock 或你自己的数据服务。3.2 数据源选择思路数据源是整套系统的地基。A 股数据可以从公开接口获取涨跌家数、成交额、指数估值港股数据相对分散需要选择覆盖港股主要指数的接口。由于接口变化较快不建议把所有接口写死在代码里而应封装成独立模块方便替换。一个比较稳妥的做法是建一个data_source.py统一导出fetch_market_snapshot()函数。后续无论切换到哪个数据提供商只需要改这一个文件。# data_source.py 示例骨架 def fetch_market_snapshot(marketA, dateNone): 返回标准化的行情快照 DataFrame字段包含: code, name, close, pct_change, turnover, pe, volume # 实际实现需要根据你的数据源接口调整 raise NotImplementedError(请替换为实际数据源实现)3.3 验证环境是否可用安装完依赖后可以先用一个最小脚本验证数据链路import akshare as ak # 尝试获取 A 股实时行情快照 # 示例代码若接口变化请以 akshare 文档为准 try: df ak.stock_zh_a_spot_em() print(A股快照获取成功共, len(df), 条记录) print(df[[代码, 名称, 最新价, 涨跌幅]].head()) except Exception as e: print(获取失败请检查数据源接口或网络环境, e)能打印出 A 股快照记录说明数据链路正常。接下来就可以做预处理和指标计算。4. 行情数据采集与预处理4.1 标准化数据字段不管从哪个数据源获取数据进入指标计算前都应该统一成以下字段字段名含义示例code证券代码600519name证券名称贵州茅台close收盘价1700.00pct_change涨跌幅百分比2.35turnover成交金额8500000000pe市盈率 TTM28.5market市场标识A / HK清洗数据时重点处理三件事去掉停牌数据、处理缺失值、过滤异常涨跌幅记录。停牌股票的涨跌幅通常为空或 0如果不剔除会拉低上涨家数比例导致温度统计失真。import pandas as pd def clean_snapshot(df: pd.DataFrame) - pd.DataFrame: 行情快照清洗 # 去停牌涨跌幅为空或成交量极小的记录剔除 df df[df[pct_change].notna()] df df[df[turnover] 0] # 去掉涨跌幅异常记录例如新上市首日无涨跌限制的品种 df df[df[pct_change].abs() 30] # 统一类型 df[pct_change] pd.to_numeric(df[pct_change], errorscoerce) df[turnover] pd.to_numeric(df[turnover], errorscoerce) df[pe] pd.to_numeric(df[pe], errorscoerce) return df新上市股票、退市整理股票往往会造成统计口径不一致。如果对结果精度要求高建议额外过滤上市不足 60 个交易日的股票。4.2 港股数据注意点港股数据相比 A 股更分散接口稳定度和字段完整性可能不足。处理港股时建议先做好三件事确认港股通标的范围是否覆盖、汇率是否统一、停牌状态能否识别。如果接口只返回部分港股需要在报告里标注“覆盖范围有限”避免直接把结果当作全市场温度。5. 温度指标计算与热度评分5.1 市场广度指标上涨家数占比市场广度是所有温度指标里最直观的一个。当上涨家数占比长期高于 70%说明市场处于普涨状态当占比低于 30%说明市场情绪偏冷。计算逻辑很简单def breadth_ratio(df: pd.DataFrame) - float: 上涨家数占比范围 0~1 up_count (df[pct_change] 0).sum() total_count len(df) return up_count / total_count if total_count 0 else 0.5如果把多个交易日的广度比例连在一起就能画出市场宽度曲线用于观察温度变化趋势。5.2 成交强度指标成交额分位数成交额体现市场的参与热度。只看单日绝对值不够因为市场总市值会变化需要用历史分位数来衡量。计算方法是取当前成交额与过去 N 个交易日的成交额序列对比得到百分位def turnover_percentile(turnover_series: pd.Series, current: float) - float: 当前成交额在历史序列中的百分位0~1 hist turnover_series.dropna().values if len(hist) 0: return 0.5 below_count (hist current).sum() return below_count / len(hist)这里的 N 建议取 250 个交易日约等于一年太长会反应迟钝太短会丢失长期参照。成交额分位数达到 0.9 以上说明当前成交量接近一年内的高位区域。5.3 估值指标市盈率百分位估值百分位是判断“泡沫”偏好在量化层面的一个重要代理指标。计算方式与成交额分位类似但要注意市盈率受盈利周期影响用 TTM 口径能平滑季节性波动。如果数据源不提供全市场市盈率可以用主要宽基指数的市盈率替代但必须在报告中注明指标口径。def pe_percentile(pe_series: pd.Series, current_pe: float) - float: 市盈率百分位0~1越高代表估值越贵 hist pe_series.dropna().values if len(hist) 0: return 0.5 below_count (hist current_pe).sum() return below_count / len(hist)正常配置下估值百分位需要先构建一个历史 PE 序列。这要求你有一个历史数据库或者每日重复抓取指数 PE 并落库。落库方案可以在后面批量任务章节展开。5.4 波动指标20 日年化波动率市场温度过高或过低常常伴随波动率异常。计算指数过去 20 个交易日收益率的年化标准差可以得到波动率指标。高波动配合高估值温度会进一步上调低波动配合低成交温度更偏向中性。import numpy as np def annualized_volatility(close_series: pd.Series, window: int 20) - float: 计算年化波动率 ret close_series.pct_change().dropna() vol ret.tail(window).std() * np.sqrt(252) return vol波动率本身不直接等于市场温度它更多作为权重因子。波动率过高时即使价格看似上涨也要给温度打折扣因为这种上涨可持续性存疑。5.5 综合温度评分把上述四个指标加权合成一个 0 到 100 的分数并映射到温度区间def market_temp_score(breadth: float, turnover_pct: float, pe_pct: float, vol: float, vol_threshold: float 0.30) - float: 综合温度评分0~100 score (breadth * 35 turnover_pct * 25 pe_pct * 25) # 波动率过高时做温和扣分 if vol vol_threshold: score - (vol - vol_threshold) * 50 score max(0, min(100, score)) return score def temp_level(score: float) - str: if score 75: return 过热 elif score 60: return 偏热 elif score 45: return 中性偏暖 elif score 30: return 中性偏冷 else: return 偏冷权重可以按需求调整但必须保持口径一致。回测时统一使用同一套权重才能让不同时间段的数据具有可比性。5.6 计算完整示例假设你已经有一个历史交易日列表和每日快照文件可以这样计算整段区间的温度序列def compute_temp_series(date_list, snapshot_loader): 计算每日温度序列 results [] for date in date_list: df snapshot_loader(date) if df is None or df.empty: continue df clean_snapshot(df) breadth breadth_ratio(df) # turnover_pct、pe_pct 需要历史序列计算 # 这里假设已有 history_turnover、history_pe 两个序列 turnover_pct turnover_percentile(history_turnover, df[turnover].sum()) pe_pct pe_percentile(history_pe, df[pe].median()) vol annualized_volatility(index_close_history) score market_temp_score(breadth, turnover_pct, pe_pct, vol) results.append({date: date, score: round(score, 2), level: temp_level(score)}) return pd.DataFrame(results)6. 批量任务与接口服务化6.1 批量回测已有历史数据批量回测的核心意义是检验当前温度在历史上的位置。假设你保存了过去三年的日频快照可以统一循环处理最终得到一张“温度历史曲线”和“当前温度分位”# 批量回测伪代码 for month in range(1, 13): snapshot load_monthly_data(2024, month) if snapshot is None: continue daily_scores compute_temp_series(snapshot[date_list], loader) daily_scores.to_csv(ftemp_history_{2024}_{month}.csv, indexFalse)这种批量任务不需要多复杂数据量也远没有图像或视频任务大。一个季度的日频快照可能只有几百 MB普通电脑完全可以处理。重点是把每次计算的输入快照和输出结果分目录保存方便后续复盘。6.2 每日自动更新温度分析如果只跑一次意义有限。更好的方式是每天收盘后自动执行一次。Windows 下可以使用计划任务Linux 下使用 Cron。先写好一个独立脚本daily_update.py# daily_update.py 示例 from datetime import datetime from data_source import fetch_market_snapshot from indicators import compute_temp_series def main(): today datetime.now().strftime(%Y-%m-%d) df fetch_market_snapshot(marketA, datetoday) score compute_temp_series([today], lambda d: df) score.to_csv(freports/temp_{today}.csv, indexFalse) print(f今日温度已更新: {today}) if __name__ __main__: main()然后配置定时任务。以下命令是 Linux Cron 示例Windows 计划任务可以在“任务计划程序”中指向同一个 Python 脚本# 每个交易日 16:30 执行一次需要替换成实际项目路径 30 16 * * 1-5 cd /path/to/project /path/to/venv/bin/python daily_update.py logs/cron.log 21执行完以后你应该能在reports目录看到当天的温度 CSV 文件。这就是最简批次自动化闭环。6.3 用 FastAPI 暴露温度接口做完批量任务后可以考虑把温度计算结果做成 HTTP 接口方便看盘工具、机器人或内部系统读取。以下代码是 FastAPI 的最小实现默认监听本机地址from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleA股港股市场温度接口) class TempResponse(BaseModel): date: str score: float level: str source: str computed_from_public_data app.get(/api/market-temp, response_modelTempResponse) def get_market_temp(date: str): # 这里应该从 CSV 或数据库读取当日计算结果 # 示例仅作占位 return TempResponse(datedate, score50.0, level中性) # 启动命令 # uvicorn main:app --host 127.0.0.1 --port 8000启动后可以用 curl 验证curl http://127.0.0.1:8000/api/market-temp?date2025-08-13如果返回 JSON 中包含score和level字段说明接口链路正常。要注意FastAPI 默认是非阻塞服务适合并发读取但在重新计算温度时不建议同时写入同一份 CSV最好用内存缓存或数据库锁。7. 资源占用与运行性能观察7.1 数据量与内存占用市场温度分析的数据量不大。全市场 A 股日频快照大约五千条记录字段十来个单个 DataFrame 内存占用在几 MB 级别。即使保存三年日频数据整体也就几百 MB。普通电脑完全跑得动关键是合理组织文件目录不要把所有历史数据一次性加载到内存。7.2 性能瓶颈在哪这个流程的瓶颈通常在数据抓取环节而不是计算环节。调用公开行情接口会受到网络延迟和数据源限流影响。如果每天拉一次快照影响不明显如果要回测几年历史数据频繁请求接口可能被限流。优化方式有三种第一次全量拉取后落盘保存后续增量更新时只拉取新增交易日数据对重复请求增加磁盘缓存或内存缓存。import os import pandas as pd CACHE_DIR cache def get_snapshot_with_cache(date: str): 带缓存的快照获取避免重复请求接口 cache_path os.path.join(CACHE_DIR, fsnapshot_{date}.csv) if os.path.exists(cache_path): return pd.read_csv(cache_path) snapshot fetch_market_snapshot_for_date(date) snapshot.to_csv(cache_path, indexFalse) return snapshot7.3 如何观察运行状态运行脚本时使用top或 Windows 任务管理器查看 Python 进程内存占用。正常情况下整个分析流程的内存占用在 200MB 以内。如果发现内存持续上涨优先检查是不是缓存目录越积越多。长时间运行时建议添加日志记录每次拉取耗时和计算耗时。8. 常见问题与排查方法问题现象可能原因排查方式解决方案接口返回数据为空数据源接口变更或网络被限制检查请求日志和返回状态码更换数据源按当前文档更新接口名涨跌家数比例异常偏高停牌股票未剔除检查快照中 pct_change 为空的数量清洗阶段过滤停牌股票市盈率缺失严重数据源对部分股票不覆盖 PE检查 pe 字段空值比例用指数 PE 替代或剔除缺失样本定时任务没有执行路径错误或虚拟环境未激活查看计划任务日志使用绝对路径在脚本中显式指定 Python 解释器接口 JSON 返回超时首次请求在拉取全量数据检查是否有缓存提前生成缓存启动时异步加载数据计算温度与直觉差异大权重设置不合理或历史序列过短检查权重和历史窗口调整权重历史窗口至少 250 个交易日FastAPI 端口被占用8000 端口已有进程监听使用 lsof / netstat 检查端口换端口启动uvicorn main:app --port 8001港股数据缺失严重接口覆盖范围不够检查返回的证券代码列表更换覆盖港股通的数据源并标注统计范围这里的排查思路是通用的不一定需要按表格逐项执行。遇到问题先看日志再看数据量最后检查是不是统计口径问题。9. 使用建议与风险边界9.1 按实际需求配置指标温度分析不是指标越多越好。建议先跑通最小集也就是“上涨家数占比 成交额分位 估值分位”等输出结果稳定后再逐步加入波动率或其他衍生指标。每加一个指标都要记录权重和影响避免系统变成不可解释的黑箱。9.2 建立目录规范项目目录建议按 input、cache、reports、logs 分四个文件夹。原始快照落到 input中间缓存落到 cache最终报告落到 reports运行状态落到 logs。这样批量任务跑半年后复盘时不会找不到数据来源。9.3 复核与合规提醒自动计算的温度结果不能直接对外发布尤其是涉及“泡沫”“过热”这类带有强主观判断的词汇时必须先由人工复核。发布报告时应注明数据来源、统计口径、覆盖范围和计算时间。涉及港股和 A 股跨市场对比时要特别说明两边数据覆盖率和币种的差异。9.4 不构成投资建议本文所有内容都属于工程技术方法介绍市场温度指标只是对公开数据的一种统计描述。任何指标都存在滞后性和幸存者偏差。不要因为“温度过热”就立刻减仓也不要因为“温度偏冷”就盲目抄底。请务必结合基本面、政策面和个人风险承受能力独立判断。10. 总结与下一步这套市场温度分析系统最适合作为复盘和数据观察的底座。先做最小集每天记录上涨家数占比、成交额分位和估值分位再结合历史区间判断当前温度位置就能把直播中经常出现的“泡沫”“热度”这类讨论变成自己可验证的数据图表。容易踩的坑集中在数据源接口不稳定、历史序列口径不一致、定时任务路径配置错误这三个地方。建议第一次先手动运行两天确认生成的 CSV 和图表符合预期再接定时任务和 FastAPI 接口。后续扩展方向也比较明确可以接入更多市场指数做横向对比可以增加行业温度拆分可以把温度评分接入自己的消息推送机器人还可以把历史温度序列与指数未来收益做相关性分析进一步检验指标的预测能力。先把基础版本跑通再逐步迭代比一开始设计大而全的系统更稳妥。建议把本文的代码框架保存下来作为市场观察的工具箱。以后看到任何人讨论“市场温度高不高”都可以用同一套口径快速算一遍用数据说话。