Python量化分析实战：构建A股港股市场情绪温度监测系统

大家好，我是专注于金融科技与市场分析的技术博主。最近，随着长鑫科技等硬核科技公司上市进程的推进，以及市场情绪的持续波动，很多开发者朋友和量化爱好者都在后台询问：如何用技术手段来感知和解析当前A股、港股市场的“温度”？传统的看盘软件指标繁多，但往往缺乏一个系统性的、可量化的“市场情绪”观测框架。

本文将从一个技术实践者的角度，手把手带你搭建一套简易但有效的市场温度监测系统。我们将不依赖任何主观判断，而是通过公开的API获取数据，使用Python进行量化分析，最终生成可视化的市场温度仪表盘。无论你是想将市场情绪因子融入自己的量化模型，还是单纯希望有一个客观的参考工具，这篇文章都能为你提供从零到一的完整代码和工程思路。

1. 市场温度解析：概念、价值与技术实现路径

在讨论“市场温度”时，我们并非指气象，而是对市场整体情绪、资金活跃度以及多空力量对比的一种量化描述。一个“高温”市场通常伴随着高成交量、普遍上涨、投资者情绪亢奋；而“低温”市场则可能成交清淡，个股涨少跌多，情绪偏向谨慎或悲观。

为什么开发者需要关注市场温度？

1. 策略择时：对于趋势跟踪或动量策略，在市场温度由冷转暖的初期介入，可能提高胜率。
2. 风险控制：当市场温度过高（过热）时，可能是短期风险积聚的信号，可据此调整仓位或增加对冲。
3. 资产配置：横跨A股、港股等多市场时，温度对比可以帮助进行资金轮动决策。
4. 产品体验：如果你在开发金融资讯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 jqdatasdk

• pandas,numpy：数据处理与分析基石。
• requests：用于调用免费或付费的金融数据API。
• matplotlib,plotly：数据可视化，plotly用于生成交互式仪表盘。
• tushare-pro：获取A股数据的主要工具，需要注册获取Token。
• jqdatasdk：聚宽数据SDK，可用于获取港股数据，同样需要注册。

重要提示：tushare-pro和jqdatasdk都是需要实名注册的合规数据源，提供了稳定、结构化的金融市场数据。请勿使用来路不明或违反数据服务协议的数据抓取方式。

2.3 数据源API配置

1. 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)

2. 聚宽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°C，50°C为中性值。

3.1 选取指标池我们为每个市场（A股、港股）选取以下基础指标，每个指标贡献一个“子温度”：

3.2 温度合成算法

1. 数据标准化：将每个指标的原始值，通过预设的映射函数，转换为0-100的子温度值。
2. 加权平均：给每个子温度分配权重，然后加权求和得到综合温度。
   • 示例权重：涨跌比(30%)， 平均涨跌幅(30%)， 成交量比率(25%)， 振幅温度(15%)。
   • 权重可根据回测结果调整。
3. 平滑处理：为避免单日数据噪声，可以对最终的综合温度进行移动平均（例如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_date=None): """ 获取A股全市场日频数据 Args: trade_date: 交易日期，格式‘YYYYMMDD’，默认为最新交易日 Returns: DataFrame: 包含股票代码、涨跌幅、成交量、振幅等 """ if trade_date is None: # 获取最近一个交易日 today = datetime.now().strftime('%Y%m%d') df_cal = pro.trade_cal(exchange='SSE', start_date=today, end_date=today) if df_cal.iloc[0]['is_open'] == 0: # 非交易日，向前找 trade_date = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d') else: trade_date = today # 获取沪深两市股票日行情 df_daily = pro.daily(trade_date=trade_date) # 获取股票基础信息，用于过滤ST、科创板/创业板等（按需） df_basic = pro.stock_basic(exchange='', list_status='L') df_merged = pd.merge(df_daily, df_basic[['ts_code', 'name']], on='ts_code', how='left') return df_merged def fetch_hk_market_data(date=None): """ 获取港股市场（恒生指数成分股）日频数据 Args: date: 日期，格式‘YYYY-MM-DD’，默认为最新交易日 Returns: DataFrame: 包含股票代码、涨跌幅、成交量、振幅等 """ if date is None: date = datetime.now().strftime('%Y-%m-%d') # 获取恒生指数成分股列表 hs_const = get_all_securities(['futures'], date=date) # 注意：这里简化，实际应取HSI成分股 # 实际应用中，应从聚宽获取准确的HSI成分股列表，此处用腾讯(00700.HK)等示例 hk_stock_list = ['00700.HK', '09988.HK', '00941.HK'] # 示例股票 # 获取行情数据 data = get_price( hk_stock_list, start_date=date, end_date=date, fields=['close', 'volume', 'high', 'low', 'pre_close'], panel=False ) 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(f"A股数据行数：{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_type='A', hist_volume_avg=None): """ 计算单一市场温度 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 result

4.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( rows=2, cols=2, 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 = "gauge+number", 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 } } ), row=1, col=1) # 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( x=indicators, y=temps, name=f'{market}股指标温度', marker_color='skyblue' if market=='A' else 'lightcoral' ), row=1, col=2 if idx==0 else 2, col=2 if idx==1 else 1) # 简化布局，实际需调整 # 3. 历史趋势图（示例，需要历史数据） # 这里假设我们有一个历史温度DataFrame `df_history` # fig.add_trace(go.Scatter(x=df_history['date'], y=df_history['A_temp'], mode='lines', name='A股'), row=2, col=2) # fig.add_trace(go.Scatter(x=df_history['date'], y=df_history['HK_temp'], mode='lines', name='港股'), row=2, col=2) fig.update_layout(height=800, title_text="A股港股市场温度实时仪表盘") 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(days=1)).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(window=5).mean().iloc[-1] if len(df_a) >=5 else df_a['volume'].mean() temp_a = calc.calculate(df_a, market_type='A', hist_volume_avg=hist_vol_avg_a) temp_hk = None if not df_hk.empty: hist_vol_avg_hk = df_hk['volume'].rolling(window=5).mean().iloc[-1] if len(df_hk) >=5 else df_hk['volume'].mean() temp_hk = calc.calculate(df_hk, market_type='HK', hist_volume_avg=hist_vol_avg_hk) # 3. 打印结果 print("\n=== 市场温度计算结果 ===") print(f"A股综合温度: {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 运行与验证

1. 在项目根目录下，确保已安装所有依赖 (pip install -r requirements.txt)。
2. 在config.py中正确配置你的Tushare和聚宽API凭证。
3. 运行主程序：python main.py。
4. 程序会输出类似以下文本结果，并弹出一个交互式的网页图表。

开始计算市场温度... 获取交易日 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. 常见问题与排查思路

在实现和运行上述系统时，你可能会遇到以下问题：

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_exists='append', index=False) conn.close()

• 错误处理与重试：网络请求必须添加超时和重试机制，并对API限流进行友好处理。

6.2 模型优化与回测

• 参数校准：本文的阈值（如涨跌比2.0为过热）是示例。应选取足够长的历史数据（如过去5年），观察指标在牛市、熊市、震荡市中的分布，以统计方法确定更合理的阈值。
• 权重优化：可以使用简单的网格搜索或更复杂的优化算法，以历史数据上温度对市场未来收益（如次日、下周涨跌）的预测能力为目标，来优化各子温度的权重。
• 引入更多指标：进阶模型可以加入：
  • 资金情绪：北向资金净流入/流出额。
  • 期权情绪：A股期权市场的认沽/认购比率（PCR）。
  • 媒体情绪：利用NLP分析财经新闻的情感倾向（需另建模型）。

6.3 系统性能与可维护性

• 模块化设计：如本文所示，将数据获取、计算逻辑、可视化分离，便于单独测试和升级。例如，可以轻松替换数据源为akshare或baostock。
• 配置化：将阈值参数、权重、API密钥等全部移至配置文件（如config.yaml）或环境变量中，避免硬编码。
• 日志记录：使用logging模块记录程序运行状态、错误信息，方便排查。

  import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logging.info(f"开始计算{trade_date}的市场温度...")

6.4 生产环境注意事项

• 合规性：确保所有数据来源的使用符合其服务条款，尤其注意数据不能用于非法商业用途或对外提供实时数据服务。
• 风险提示：任何量化模型都是对历史的总结和归纳，市场温度仅作为辅助参考工具，不能作为唯一的投资决策依据。务必理解模型的局限性。
• 服务降级：当某个数据源（如港股API）失效时，系统应能降级运行，继续计算A股温度，而不是整体崩溃。

通过本文的教程，你不仅获得了一个可运行的市场温度计代码，更重要的是掌握了一套从数据获取、指标构建、模型计算到可视化展示的完整量化分析流程。这套方法可以迁移到任何你感兴趣的市场分析场景中，例如板块温度、个股情绪、债市情绪等。

技术的价值在于将模糊的感知变为清晰的度量。下一步，你可以尝试将此温度指标与简单的择时策略结合进行回测，或者将其作为一个特征输入到更复杂的机器学习模型中。记住，持续迭代和验证是量化工作的核心。