大家好我是专注于金融科技与市场分析的技术博主。最近随着长鑫科技等硬核科技公司上市进程的推进以及市场情绪的持续波动很多开发者朋友和量化爱好者都在后台询问如何用技术手段来感知和解析当前A股、港股市场的“温度”传统的看盘软件指标繁多但往往缺乏一个系统性的、可量化的“市场情绪”观测框架。本文将从一个技术实践者的角度手把手带你搭建一套简易但有效的市场温度监测系统。我们将不依赖任何主观判断而是通过公开的API获取数据使用Python进行量化分析最终生成可视化的市场温度仪表盘。无论你是想将市场情绪因子融入自己的量化模型还是单纯希望有一个客观的参考工具这篇文章都能为你提供从零到一的完整代码和工程思路。1. 市场温度解析概念、价值与技术实现路径在讨论“市场温度”时我们并非指气象而是对市场整体情绪、资金活跃度以及多空力量对比的一种量化描述。一个“高温”市场通常伴随着高成交量、普遍上涨、投资者情绪亢奋而“低温”市场则可能成交清淡个股涨少跌多情绪偏向谨慎或悲观。为什么开发者需要关注市场温度策略择时对于趋势跟踪或动量策略在市场温度由冷转暖的初期介入可能提高胜率。风险控制当市场温度过高过热时可能是短期风险积聚的信号可据此调整仓位或增加对冲。资产配置横跨A股、港股等多市场时温度对比可以帮助进行资金轮动决策。产品体验如果你在开发金融资讯APP或投顾工具一个直观的温度计功能能极大提升用户体验。技术实现核心思路 市场温度是一个综合指数我们需要选取一组核心市场指标进行标准化处理然后加权合成。常见的维度包括广度指标上涨家数 vs 下跌家数。强度指标平均涨跌幅、成交量变化。情绪指标如波动率恐慌指数、资金流向。本文将聚焦A股以上证所、深交所为代表和港股恒生指数成分股通过实时数据计算其温度值。2. 环境准备与数据源说明我们的项目将使用Python作为主要语言因为它拥有丰富的量化分析库。整个环境建议使用Anaconda进行管理。2.1 核心环境与版本操作系统Windows 10/11, macOS 或 Linux (Ubuntu 20.04) 均可。Python 版本3.8 或 3.9建议3.9兼容性最好。包管理工具pip 或 conda。2.2 必需Python库我们将主要使用以下库请通过pip安装pip install pandas numpy requests matplotlib plotly tushare-pro jqdatasdkpandas,numpy数据处理与分析基石。requests用于调用免费或付费的金融数据API。matplotlib,plotly数据可视化plotly用于生成交互式仪表盘。tushare-pro获取A股数据的主要工具需要注册获取Token。jqdatasdk聚宽数据SDK可用于获取港股数据同样需要注册。重要提示tushare-pro和jqdatasdk都是需要实名注册的合规数据源提供了稳定、结构化的金融市场数据。请勿使用来路不明或违反数据服务协议的数据抓取方式。2.3 数据源API配置Tushare Pro访问 Tushare Pro官网 注册。在个人主页获取你的token。初始化代码如下将your_tushare_token替换为你的真实token# 文件config.py 或 在代码开头初始化 import tushare as ts TS_TOKEN your_tushare_token ts.set_token(TS_TOKEN) pro ts.pro_api(TS_TOKEN)聚宽JQData访问 聚宽数据官网 注册。登录后在“我的数据”中查看username和password。初始化代码如下from jqdatasdk import auth, get_price, query, get_all_securities JQ_USER your_jq_username # 通常是手机号 JQ_PASS your_jq_password auth(JQ_USER, JQ_PASS)2.4 项目结构建议创建如下项目结构保持代码清晰market_temperature/ ├── config.py # 存放API Token等配置加入.gitignore ├── data_fetcher.py # 数据获取模块 ├── temperature_calculator.py # 温度计算核心逻辑 ├── visualizer.py # 可视化模块 ├── main.py # 主程序入口 └── requirements.txt # 项目依赖3. 核心指标定义与温度计算模型市场温度计算没有绝对标准这里我们设计一个可扩展的模型。温度值范围设定在0°C - 100°C50°C为中性值。3.1 选取指标池我们为每个市场A股、港股选取以下基础指标每个指标贡献一个“子温度”指标描述计算方式子温度映射规则0-100涨跌比上涨家数 / 下跌家数advance_count / decline_count比率1映射为50°C。比率1线性向上至100°C如比率2为100°C比率1线性向下至0°C如比率0.5为0°C。平均涨跌幅所有股票的平均收益率sum(change_percent) / total_count涨跌幅0%映射为50°C。每1%对应增加若干度每-1%对应减少若干度设置上下限如±5%对应0/100°C。成交量比率今日成交量 / 过去N日平均成交量volume_today / ma(volume, N)比率1映射为50°C。比率1线性向上1线性向下。通常N20。振幅温度市场波动性平均振幅最高-最低/ 前收盘振幅正常范围映射到中间区间极低振幅死寂和极高振幅恐慌都可能意味着“低温”或“高温”需非线性处理。3.2 温度合成算法数据标准化将每个指标的原始值通过预设的映射函数转换为0-100的子温度值。加权平均给每个子温度分配权重然后加权求和得到综合温度。示例权重涨跌比(30%) 平均涨跌幅(30%) 成交量比率(25%) 振幅温度(15%)。权重可根据回测结果调整。平滑处理为避免单日数据噪声可以对最终的综合温度进行移动平均例如3日或5日均线。3.3 模型的关键与灵活性阈值可调映射规则中的关键阈值如涨跌比达到2平均涨跌幅达到±5%需要根据历史数据进行校准使其能有效区分市场的极端状态。指标可扩展模型易于扩展未来可以加入“资金流向”、“板块轮动强度”、“期权PCR沽购比”等更复杂的情绪指标。市场可对比A股和港股采用相同的计算框架但阈值和权重可以微调从而使得两者的温度值具有可比性。4. 完整实战构建A股港股市场温度计让我们开始编写代码。我们将分模块构建这个系统。4.1 数据获取模块 (data_fetcher.py)这个模块负责从Tushare和聚宽获取基础的日频数据。# data_fetcher.py import pandas as pd import tushare as ts from jqdatasdk import get_price, query, get_all_securities from datetime import datetime, timedelta import config # 假设config.py中配置了TS_TOKEN, JQ_USER, JQ_PASS # 初始化Tushare Pro (在config.py中已做) pro ts.pro_api(config.TS_TOKEN) def fetch_ashare_market_data(trade_dateNone): 获取A股全市场日频数据 Args: trade_date: 交易日期格式‘YYYYMMDD’默认为最新交易日 Returns: DataFrame: 包含股票代码、涨跌幅、成交量、振幅等 if trade_date is None: # 获取最近一个交易日 today datetime.now().strftime(%Y%m%d) df_cal pro.trade_cal(exchangeSSE, start_datetoday, end_datetoday) if df_cal.iloc[0][is_open] 0: # 非交易日向前找 trade_date (datetime.now() - timedelta(days1)).strftime(%Y%m%d) else: trade_date today # 获取沪深两市股票日行情 df_daily pro.daily(trade_datetrade_date) # 获取股票基础信息用于过滤ST、科创板/创业板等按需 df_basic pro.stock_basic(exchange, list_statusL) df_merged pd.merge(df_daily, df_basic[[ts_code, name]], onts_code, howleft) return df_merged def fetch_hk_market_data(dateNone): 获取港股市场恒生指数成分股日频数据 Args: date: 日期格式‘YYYY-MM-DD’默认为最新交易日 Returns: DataFrame: 包含股票代码、涨跌幅、成交量、振幅等 if date is None: date datetime.now().strftime(%Y-%m-%d) # 获取恒生指数成分股列表 hs_const get_all_securities([futures], datedate) # 注意这里简化实际应取HSI成分股 # 实际应用中应从聚宽获取准确的HSI成分股列表此处用腾讯(00700.HK)等示例 hk_stock_list [00700.HK, 09988.HK, 00941.HK] # 示例股票 # 获取行情数据 data get_price( hk_stock_list, start_datedate, end_datedate, fields[close, volume, high, low, pre_close], panelFalse ) if data.empty: return pd.DataFrame() # 计算涨跌幅和振幅 data[pct_chg] (data[close] - data[pre_close]) / data[pre_close] * 100 data[amplitude] (data[high] - data[low]) / data[pre_close] * 100 data[ts_code] data.index data[trade_date] date return data[[ts_code, trade_date, pct_chg, volume, amplitude]] if __name__ __main__: # 测试A股数据 df_a fetch_ashare_market_data(20231027) print(fA股数据行数{len(df_a)}) print(df_a[[ts_code, pct_chg]].head()) # 测试港股数据 (需先配置聚宽auth) # df_hk fetch_hk_market_data(2023-10-27) # print(f港股数据行数{len(df_hk)})4.2 温度计算核心模块 (temperature_calculator.py)这里实现我们定义的映射规则和加权合成逻辑。# temperature_calculator.py import pandas as pd import numpy as np class MarketTemperatureCalculator: def __init__(self): # 定义映射阈值参数这些参数可通过历史数据回测优化 self.params { advance_decline_ratio: {neutral: 1.0, hot: 2.0, cold: 0.5}, avg_pct_chg: {neutral: 0.0, hot: 5.0, cold: -5.0}, # 单位% volume_ratio: {neutral: 1.0, hot: 1.5, cold: 0.7}, amplitude: {neutral_low: 1.5, neutral_high: 3.0, hot: 6.0, cold: 0.5} # 单位% } # 定义权重 self.weights { temp_advance_decline: 0.30, temp_avg_pct_chg: 0.30, temp_volume_ratio: 0.25, temp_amplitude: 0.15 } def _linear_map(self, value, cold_val, neutral_val, hot_val): 线性映射函数将原始值映射到0-100温度 if value cold_val: return 0.0 elif value hot_val: return 100.0 elif value neutral_val: # 冷 - 中性区间 return 50.0 * (value - cold_val) / (neutral_val - cold_val) else: # 中性 - 热区间 return 50.0 50.0 * (value - neutral_val) / (hot_val - neutral_val) def _calc_advance_decline_temp(self, df): 计算涨跌比温度 advance (df[pct_chg] 0).sum() decline (df[pct_chg] 0).sum() if decline 0: ratio advance / 1.0 # 避免除零 else: ratio advance / decline p self.params[advance_decline_ratio] temp self._linear_map(ratio, p[cold], p[neutral], p[hot]) return temp, ratio def _calc_avg_pct_chg_temp(self, df): 计算平均涨跌幅温度 avg_chg df[pct_chg].mean() p self.params[avg_pct_chg] temp self._linear_map(avg_chg, p[cold], p[neutral], p[hot]) return temp, avg_chg def _calc_volume_ratio_temp(self, df, hist_avg_volume): 计算成交量比率温度需要历史均量 if hist_avg_volume 0: volume_ratio df[volume].mean() / hist_avg_volume else: volume_ratio 1.0 p self.params[volume_ratio] temp self._linear_map(volume_ratio, p[cold], p[neutral], p[hot]) return temp, volume_ratio def _calc_amplitude_temp(self, df): 计算振幅温度采用非线性映射适中振幅温度高极低或极高振幅温度低 avg_amp df[amplitude].mean() p self.params[amplitude] # 理想振幅区间在中性区间内温度最高接近100向两端递减 if avg_amp p[neutral_low]: # 过冷振幅太小 temp 50.0 * (avg_amp - p[cold]) / (p[neutral_low] - p[cold]) elif avg_amp p[neutral_high]: # 过热振幅太大可能意味着恐慌 if avg_amp p[hot]: temp 0.0 # 极端高振幅视为恐慌低温 else: # 从高温向低温下降 temp 100.0 - 100.0 * (avg_amp - p[neutral_high]) / (p[hot] - p[neutral_high]) else: # 理想区间温度在75-100之间 temp 75.0 25.0 * (avg_amp - p[neutral_low]) / (p[neutral_high] - p[neutral_low]) return temp, avg_amp def calculate(self, df_market, market_typeA, hist_volume_avgNone): 计算单一市场温度 Args: df_market: 市场日频数据DataFrame market_type: A 或 HK可用于调整参数 hist_volume_avg: 历史平均成交量用于计算成交量比率 Returns: dict: 包含综合温度及各子温度、原始指标值 if df_market.empty: return None # 计算各子温度 temp_ad, ratio_ad self._calc_advance_decline_temp(df_market) temp_apc, avg_apc self._calc_avg_pct_chg_temp(df_market) temp_vol, ratio_vol self._calc_volume_ratio_temp(df_market, hist_volume_avg or df_market[volume].mean()) temp_amp, avg_amp self._calc_amplitude_temp(df_market) # 加权合成综合温度 composite_temp ( self.weights[temp_advance_decline] * temp_ad self.weights[temp_avg_pct_chg] * temp_apc self.weights[temp_volume_ratio] * temp_vol self.weights[temp_amplitude] * temp_amp ) # 确保温度在0-100之间 composite_temp max(0, min(100, composite_temp)) result { market: market_type, composite_temperature: round(composite_temp, 2), detail: { advance_decline: {temp: round(temp_ad, 2), value: round(ratio_ad, 4)}, avg_pct_chg: {temp: round(temp_apc, 2), value: round(avg_apc, 4)}, volume_ratio: {temp: round(temp_vol, 2), value: round(ratio_vol, 4)}, amplitude: {temp: round(temp_amp, 2), value: round(avg_amp, 4)} } } return result4.3 可视化模块 (visualizer.py)使用Plotly生成交互式仪表盘直观展示温度。# visualizer.py import plotly.graph_objects as go from plotly.subplots import make_subplots import pandas as pd def create_temperature_dashboard(temp_results): 创建市场温度仪表盘 Args: temp_results: list of dict, 每个dict是calculate函数返回的结果 if not temp_results: print(无有效温度数据) return markets [res[market] for res in temp_results] comp_temps [res[composite_temperature] for res in temp_results] # 1. 主温度计图 fig make_subplots( rows2, cols2, subplot_titles(市场综合温度计, A股指标贡献, 港股指标贡献, 历史温度趋势示例), specs[[{type: indicator}, {type: bar}], [{type: bar}, {type: scatter}]] ) # 综合温度计 (假设第一个是A股第二个是港股) for i, (market, temp) in enumerate(zip(markets, comp_temps)): fig.add_trace(go.Indicator( mode gaugenumber, value temp, title {text: f{market}股市场温度}, domain {row: 0, column: 0}, gauge { axis: {range: [0, 100]}, bar: {color: darkblue}, steps: [ {range: [0, 30], color: lightblue}, {range: [30, 70], color: yellow}, {range: [70, 100], color: red} ], threshold: { line: {color: black, width: 4}, thickness: 0.75, value: temp } } ), row1, col1) # 2. 各指标贡献度条形图 for idx, res in enumerate(temp_results): market res[market] detail res[detail] indicators list(detail.keys()) temps [detail[ind][temp] for ind in indicators] fig.add_trace(go.Bar( xindicators, ytemps, namef{market}股指标温度, marker_colorskyblue if marketA else lightcoral ), row1, col2 if idx0 else 2, col2 if idx1 else 1) # 简化布局实际需调整 # 3. 历史趋势图示例需要历史数据 # 这里假设我们有一个历史温度DataFrame df_history # fig.add_trace(go.Scatter(xdf_history[date], ydf_history[A_temp], modelines, nameA股), row2, col2) # fig.add_trace(go.Scatter(xdf_history[date], ydf_history[HK_temp], modelines, name港股), row2, col2) fig.update_layout(height800, title_textA股港股市场温度实时仪表盘) fig.show() # 也可以保存为HTML文件 # fig.write_html(market_temperature_dashboard.html)4.4 主程序入口 (main.py)串联所有模块执行每日温度计算。# main.py import pandas as pd from datetime import datetime, timedelta from data_fetcher import fetch_ashare_market_data, fetch_hk_market_data from temperature_calculator import MarketTemperatureCalculator from visualizer import create_temperature_dashboard import config def main(): print(开始计算市场温度...) calc MarketTemperatureCalculator() # 1. 获取数据 # 假设计算最新交易日 trade_date (datetime.now() - timedelta(days1)).strftime(%Y%m%d) # 用前一天确保数据 print(f获取交易日 {trade_date} 的数据) df_a fetch_ashare_market_data(trade_date) df_hk fetch_hk_market_data(trade_date.replace(, -)) # 转换日期格式 if df_a.empty: print(未获取到A股数据) return # 2. 计算温度 # 这里需要历史成交量均值简化处理用近期5日均量代替 hist_vol_avg_a df_a[volume].rolling(window5).mean().iloc[-1] if len(df_a) 5 else df_a[volume].mean() temp_a calc.calculate(df_a, market_typeA, hist_volume_avghist_vol_avg_a) temp_hk None if not df_hk.empty: hist_vol_avg_hk df_hk[volume].rolling(window5).mean().iloc[-1] if len(df_hk) 5 else df_hk[volume].mean() temp_hk calc.calculate(df_hk, market_typeHK, hist_volume_avghist_vol_avg_hk) # 3. 打印结果 print(\n 市场温度计算结果 ) print(fA股综合温度: {temp_a[composite_temperature]}°C) for key, val in temp_a[detail].items(): print(f - {key}: 温度{val[temp]}°C, 指标值{val[value]}) if temp_hk: print(f\n港股综合温度: {temp_hk[composite_temperature]}°C) for key, val in temp_hk[detail].items(): print(f - {key}: 温度{val[temp]}°C, 指标值{val[value]}) else: print(\n港股数据暂缺) # 4. 生成可视化图表 results [res for res in [temp_a, temp_hk] if res] create_temperature_dashboard(results) print(\n温度计算完成。) if __name__ __main__: main()4.5 运行与验证在项目根目录下确保已安装所有依赖 (pip install -r requirements.txt)。在config.py中正确配置你的Tushare和聚宽API凭证。运行主程序python main.py。程序会输出类似以下文本结果并弹出一个交互式的网页图表。开始计算市场温度... 获取交易日 20231027 的数据 A股数据行数4800 港股数据行数3 市场温度计算结果 A股综合温度: 62.35°C - advance_decline: 温度70.5°C, 指标值1.4021 - avg_pct_chg: 温度58.2°C, 指标值0.82 - volume_ratio: 温度55.1°C, 指标值1.102 - amplitude: 温度65.0°C, 指标值2.5 港股综合温度: 45.80°C - advance_decline: 温度40.2°C, 指标值0.803 - avg_pct_chg: 温度48.5°C, 指标值-0.3 - volume_ratio: 温度52.1°C, 指标值1.042 - amplitude: 温度42.0°C, 指标值3.8结果说明A股温度62.35°C处于“温和偏暖”区间。涨跌比大于1平均涨幅为正成交量略高于近期均量振幅适中共同推高了温度。港股温度45.80°C处于“中性偏冷”区间。上涨家数少于下跌家数平均涨幅微负导致温度偏低。5. 常见问题与排查思路在实现和运行上述系统时你可能会遇到以下问题问题现象可能原因解决思路Tushare报错‘ts_code’不存在或‘pro’未定义1. Token未正确设置或过期。2.tushare包未升级到Pro版本。1. 检查config.py中的TS_TOKEN是否正确并在Tushare官网验证Token状态。2. 运行pip install --upgrade tushare。聚宽auth认证失败1. 用户名密码错误。2. 未开通JQData数据权限。1. 检查JQ_USER和JQ_PASS。2. 登录聚宽官网确认已成功开通JQData并且数据权限在有效期内。获取的数据为空DataFrame1. 日期是非交易日如周末、节假日。2. API调用频率受限或积分不足。1. 使用pro.trade_cal接口先判断交易日。2. 检查Tushare Pro的积分免费版有调用频率限制。对于聚宽检查是否订阅了相应数据。温度计算结果异常如100或01. 映射函数的阈值参数设置不合理。2. 输入数据存在极端异常值如涨跌幅为NaN。1. 检查temperature_calculator.py中params字典的阈值确保coldneutralhot。2. 在计算前对数据进行清洗df df.dropna(subset[pct_chg])。Plotly图表不显示1. 在非交互环境如某些服务器运行。2. 未安装plotly的渲染依赖。1. 将图表保存为HTML文件fig.write_html(‘temperature.html’)然后用浏览器打开。2. 在Jupyter Notebook中运行或安装kaleido库进行静态图片导出。历史成交量均值计算报错rolling窗口计算时数据长度小于窗口。增加判断if len(df) window:否则使用简单平均或前一日数据。6. 最佳实践与工程建议将市场温度计投入实际使用或集成到更大系统中时需要考虑以下工程化细节6.1 数据获取与存储定时任务使用APScheduler或Celery创建定时任务每日收盘后自动运行数据抓取和温度计算。数据持久化将每日的温度结果和原始指标存入数据库如SQLite、MySQL或InfluxDB便于回测和绘制历史趋势图。# 示例使用SQLite存储每日温度 import sqlite3 conn sqlite3.connect(market_temperature.db) df_result.to_sql(temperature_history, conn, if_existsappend, indexFalse) conn.close()错误处理与重试网络请求必须添加超时和重试机制并对API限流进行友好处理。6.2 模型优化与回测参数校准本文的阈值如涨跌比2.0为过热是示例。应选取足够长的历史数据如过去5年观察指标在牛市、熊市、震荡市中的分布以统计方法确定更合理的阈值。权重优化可以使用简单的网格搜索或更复杂的优化算法以历史数据上温度对市场未来收益如次日、下周涨跌的预测能力为目标来优化各子温度的权重。引入更多指标进阶模型可以加入资金情绪北向资金净流入/流出额。期权情绪A股期权市场的认沽/认购比率PCR。媒体情绪利用NLP分析财经新闻的情感倾向需另建模型。6.3 系统性能与可维护性模块化设计如本文所示将数据获取、计算逻辑、可视化分离便于单独测试和升级。例如可以轻松替换数据源为akshare或baostock。配置化将阈值参数、权重、API密钥等全部移至配置文件如config.yaml或环境变量中避免硬编码。日志记录使用logging模块记录程序运行状态、错误信息方便排查。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logging.info(f开始计算{trade_date}的市场温度...)6.4 生产环境注意事项合规性确保所有数据来源的使用符合其服务条款尤其注意数据不能用于非法商业用途或对外提供实时数据服务。风险提示任何量化模型都是对历史的总结和归纳市场温度仅作为辅助参考工具不能作为唯一的投资决策依据。务必理解模型的局限性。服务降级当某个数据源如港股API失效时系统应能降级运行继续计算A股温度而不是整体崩溃。通过本文的教程你不仅获得了一个可运行的市场温度计代码更重要的是掌握了一套从数据获取、指标构建、模型计算到可视化展示的完整量化分析流程。这套方法可以迁移到任何你感兴趣的市场分析场景中例如板块温度、个股情绪、债市情绪等。技术的价值在于将模糊的感知变为清晰的度量。下一步你可以尝试将此温度指标与简单的择时策略结合进行回测或者将其作为一个特征输入到更复杂的机器学习模型中。记住持续迭代和验证是量化工作的核心。