用Python处理腾讯股票API分时数据：手把手教你计算均价线（附完整代码）
·
Python实战：腾讯股票API分时数据均价线计算全解析
在量化交易和股票数据分析中，分时图是最基础也最重要的图表之一。而均价线作为分时图的核心指标，能够直观反映股票在特定时间段内的平均成本分布。本文将手把手教你如何利用Python处理腾讯股票API返回的分时数据，从原始数据清洗到均价线计算，最终实现可视化呈现的全过程。
1. 理解分时数据与均价线原理
1.1 分时数据结构解析
腾讯股票API返回的分时数据通常采用以下格式：
{
"code": 0,
"msg": "",
"data": {
"sh600519": {
"data": {
"data": [
"0930 2000.00 925",
"0931 1981.01 1321",
"0932 1984.88 1754"
],
"date": "20210317"
}
}
}
}
每条数据包含三个关键字段：
- 时间 ：如"0930"表示9:30
- 当前价格 ：该分钟的最后成交价
- 累计成交量 ：从开盘到该分钟的总成交量
1.2 均价线计算公式推导
均价线的计算需要理解以下核心概念：
- 当前分钟成交量 = 当前累计成交量 - 上一分钟累计成交量
- 当前分钟成交额 = 当前价格 × 当前分钟成交量
- 累计成交额 = ∑(各分钟成交额)
- 均价 = 累计成交额 ÷ 当前累计成交量
以示例数据中的前两分钟为例：
| 时间 | 价格 | 累计成交量 | 计算过程 | 均价结果 |
|---|---|---|---|---|
| 0930 | 2000.00 | 925 | (2000×925)/925 | 2000.00 |
| 0931 | 1981.01 | 1321 | (2000×925 + 1981.01×396)/1321 | 1994.31 |
2. 数据预处理与清洗
2.1 解析原始JSON数据
首先我们需要将API返回的JSON数据转换为可操作的Python数据结构：
import json
from typing import List, Dict
def parse_raw_data(raw_json: str) -> Dict:
"""
解析腾讯API返回的原始JSON数据
返回格式: {'date': '20210317', 'data': [['0930', 2000.00, 925], ...]}
"""
parsed = json.loads(raw_json)
stock_data = parsed['data']['sh600519']['data']
result = {
'date': stock_data['date'],
'data': []
}
for item in stock_data['data']:
time, price, volume = item.split()
result['data'].append([
time,
float(price),
int(volume)
])
return result
2.2 数据有效性验证
处理金融数据时，数据质量至关重要。我们需要添加验证逻辑：
def validate_data(data: List) -> bool:
"""
验证分时数据的有效性
"""
if not data:
return False
# 检查时间序列是否连续
prev_time = data[0][0]
for item in data[1:]:
current_time = item[0]
# 简单验证时间是否递增
if current_time <= prev_time:
return False
prev_time = current_time
# 检查成交量为非递减
prev_volume = data[0][2]
for item in data[1:]:
if item[2] < prev_volume:
return False
prev_volume = item[2]
return True
3. 核心计算逻辑实现
3.1 计算每分钟成交量
基于累计成交量计算每分钟的实际成交量：
def calculate_minute_volume(data: List) -> List:
"""
计算每分钟成交量
返回新增volume字段: [time, price, cum_volume, minute_volume]
"""
if not data:
return []
result = []
prev_volume = 0
for item in data:
time, price, cum_volume = item
minute_volume = cum_volume - prev_volume
# 处理第一分钟特殊情况
if prev_volume == 0:
minute_volume = cum_volume
result.append([time, price, cum_volume, minute_volume])
prev_volume = cum_volume
return result
3.2 实现均价线计算
完整实现均价线计算逻辑：
def calculate_average_price(data: List) -> List:
"""
计算每分钟的均价线
返回格式: [time, price, cum_volume, minute_volume, average_price]
"""
if not data:
return []
data_with_volume = calculate_minute_volume(data)
result = []
cumulative_amount = 0.0
cumulative_volume = 0
for item in data_with_volume:
time, price, cum_volume, minute_volume = item
# 计算累计成交额
minute_amount = price * minute_volume
cumulative_amount += minute_amount
cumulative_volume += minute_volume
# 计算均价并四舍五入到2位小数
if cumulative_volume == 0:
average_price = price
else:
average_price = round(cumulative_amount / cumulative_volume, 2)
result.append([time, price, cum_volume, minute_volume, average_price])
return result
4. 数据可视化与分析
4.1 使用Matplotlib绘制分时图
将计算好的数据可视化呈现：
import matplotlib.pyplot as plt
from matplotlib.dates import DateFormatter, HourLocator
import datetime
def plot_time_series(data: List, title: str = "分时图"):
"""
绘制分时图与均价线
"""
times = []
prices = []
averages = []
for item in data:
time_str = item[0]
# 将"0930"转换为datetime时间对象
time_obj = datetime.datetime.strptime(f"2021-01-01 {time_str[:2]}:{time_str[2:]}:00", "%Y-%m-%d %H:%M:%S")
times.append(time_obj)
prices.append(item[1])
averages.append(item[4])
plt.figure(figsize=(12, 6))
plt.plot(times, prices, label='当前价格', color='#1f77b4')
plt.plot(times, averages, label='均价线', color='#ff7f0e', linestyle='--')
# 设置x轴格式
ax = plt.gca()
ax.xaxis.set_major_locator(HourLocator(interval=1))
ax.xaxis.set_major_formatter(DateFormatter("%H:%M"))
plt.title(title)
plt.xlabel("时间")
plt.ylabel("价格")
plt.legend()
plt.grid(True, linestyle='--', alpha=0.7)
plt.tight_layout()
plt.show()
4.2 完整流程示例
将上述功能整合成完整的工作流程：
# 示例数据
sample_json = """
{
"code": 0,
"msg": "",
"data": {
"sh600519": {
"data": {
"data": [
"0930 2000.00 925",
"0931 1981.01 1321",
"0932 1984.88 1754",
"0933 1980.03 2033",
"0934 1988.98 2243"
],
"date": "20210317"
}
}
}
}
"""
# 执行完整流程
parsed_data = parse_raw_data(sample_json)
if validate_data(parsed_data['data']):
calculated_data = calculate_average_price(parsed_data['data'])
plot_time_series(calculated_data, "贵州茅台分时图")
else:
print("数据验证失败，请检查数据质量")
5. 高级应用与优化
5.1 处理异常数据情况
实际应用中需要考虑各种异常情况：
def safe_calculate_average_price(data: List) -> List:
"""
带异常处理的均价计算
"""
if not data:
return []
result = []
cumulative_amount = 0.0
cumulative_volume = 0
prev_volume = 0
for i, item in enumerate(data):
try:
time, price, cum_volume = item
# 验证数据有效性
if not (isinstance(price, (int, float)) and isinstance(cum_volume, int)):
raise ValueError(f"Invalid data type at index {i}")
if cum_volume < prev_volume and i > 0:
raise ValueError(f"Decreasing volume at index {i}")
minute_volume = cum_volume - prev_volume if i > 0 else cum_volume
if minute_volume < 0:
raise ValueError(f"Negative minute volume at index {i}")
# 计算成交额和均价
minute_amount = price * minute_volume
cumulative_amount += minute_amount
cumulative_volume += minute_volume
average_price = round(cumulative_amount / cumulative_volume, 2) if cumulative_volume > 0 else price
result.append([time, price, cum_volume, minute_volume, average_price])
prev_volume = cum_volume
except Exception as e:
print(f"Error processing data at index {i}: {str(e)}")
# 可以选择跳过错误数据或使用前值填充
if result: # 使用前一个有效值
last_valid = result[-1]
result.append([time, last_valid[1], cum_volume, 0, last_valid[4]])
else: # 没有前值可用
result.append([time, price, cum_volume, 0, price])
return result
5.2 性能优化建议
当处理大量股票或长时间序列数据时，性能优化很重要：
- 使用NumPy向量化计算 ：
import numpy as np
def vectorized_calculation(data: List):
"""使用NumPy进行向量化计算"""
arr = np.array(data)
times = arr[:, 0]
prices = arr[:, 1].astype(float)
cum_volumes = arr[:, 2].astype(int)
# 计算分钟成交量
minute_volumes = np.zeros_like(cum_volumes, dtype=float)
minute_volumes[1:] = np.diff(cum_volumes)
minute_volumes[0] = cum_volumes[0]
# 计算累计成交额
minute_amounts = prices * minute_volumes
cumulative_amounts = np.cumsum(minute_amounts)
cumulative_volumes = np.cumsum(minute_volumes)
# 计算均价
average_prices = np.round(cumulative_amounts / cumulative_volumes, 2)
return list(zip(times, prices, cum_volumes, minute_volumes, average_prices))
-
多线程处理 ：当需要处理多只股票数据时，可以使用Python的
concurrent.futures实现并行计算。 -
缓存中间结果 ：将计算好的均价数据存储起来，避免重复计算。
5.3 扩展应用：交易信号生成
基于均价线可以开发简单的交易策略：
def generate_signals(data: List) -> List:
"""
基于价格与均价线关系生成交易信号
返回: [time, price, average, signal]
signal: 1=买入, -1=卖出, 0=持有
"""
if len(data) < 2:
return []
signals = []
signals.append([data[0][0], data[0][1], data[0][4], 0]) # 第一个点无信号
for i in range(1, len(data)):
prev_price = data[i-1][1]
prev_avg = data[i-1][4]
current_price = data[i][1]
current_avg = data[i][4]
# 简单策略：价格上穿均价线买入，下穿卖出
if prev_price < prev_avg and current_price > current_avg:
signal = 1 # 买入信号
elif prev_price > prev_avg and current_price < current_avg:
signal = -1 # 卖出信号
else:
signal = 0 # 无信号
signals.append([data[i][0], current_price, current_avg, signal])
return signals
6. 完整代码整合
将所有功能整合为一个完整的Python类：
import json
import datetime
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.dates import DateFormatter, HourLocator
from typing import List, Dict, Optional
class TencentStockAnalyzer:
def __init__(self):
self.raw_data = None
self.parsed_data = None
self.calculated_data = None
def load_from_json(self, json_str: str) -> bool:
"""加载并解析JSON数据"""
try:
self.raw_data = json.loads(json_str)
stock_code = next(iter(self.raw_data['data'].keys()))
stock_data = self.raw_data['data'][stock_code]['data']
self.parsed_data = {
'code': stock_code,
'date': stock_data['date'],
'data': []
}
for item in stock_data['data']:
time, price, volume = item.split()
self.parsed_data['data'].append([
time,
float(price),
int(volume)
])
return True
except Exception as e:
print(f"Error parsing JSON: {str(e)}")
return False
def calculate_averages(self) -> Optional[List]:
"""计算均价线数据"""
if not self.parsed_data or not self.validate_data():
return None
data = self.parsed_data['data']
arr = np.array(data)
times = arr[:, 0]
prices = arr[:, 1].astype(float)
cum_volumes = arr[:, 2].astype(int)
# 向量化计算
minute_volumes = np.zeros_like(cum_volumes, dtype=float)
minute_volumes[1:] = np.diff(cum_volumes)
minute_volumes[0] = cum_volumes[0]
minute_amounts = prices * minute_volumes
cumulative_amounts = np.cumsum(minute_amounts)
cumulative_volumes = np.cumsum(minute_volumes)
average_prices = np.round(cumulative_amounts / cumulative_volumes, 2)
self.calculated_data = list(zip(
times, prices, cum_volumes, minute_volumes, average_prices
))
return self.calculated_data
def validate_data(self) -> bool:
"""验证数据有效性"""
if not self.parsed_data or not self.parsed_data['data']:
return False
data = self.parsed_data['data']
# 检查时间序列是否连续
prev_time = data[0][0]
for item in data[1:]:
if item[0] <= prev_time:
return False
prev_time = item[0]
# 检查成交量为非递减
prev_volume = data[0][2]
for item in data[1:]:
if item[2] < prev_volume:
return False
prev_volume = item[2]
return True
def plot_chart(self, title: str = None) -> None:
"""绘制分时图"""
if not self.calculated_data:
print("No calculated data available")
return
if not title:
title = f"{self.parsed_data['code']} 分时图 - {self.parsed_data['date']}"
times = []
prices = []
averages = []
for item in self.calculated_data:
time_str = item[0]
time_obj = datetime.datetime.strptime(
f"2021-01-01 {time_str[:2]}:{time_str[2:]}:00",
"%Y-%m-%d %H:%M:%S"
)
times.append(time_obj)
prices.append(item[1])
averages.append(item[4])
plt.figure(figsize=(12, 6))
plt.plot(times, prices, label='当前价格', color='#1f77b4')
plt.plot(times, averages, label='均价线', color='#ff7f0e', linestyle='--')
ax = plt.gca()
ax.xaxis.set_major_locator(HourLocator(interval=1))
ax.xaxis.set_major_formatter(DateFormatter("%H:%M"))
plt.title(title)
plt.xlabel("时间")
plt.ylabel("价格")
plt.legend()
plt.grid(True, linestyle='--', alpha=0.7)
plt.tight_layout()
plt.show()
# 使用示例
if __name__ == "__main__":
analyzer = TencentStockAnalyzer()
# 替换为实际API返回的JSON数据
sample_json = """{"code":0,"msg":"","data":{"sh600519":{"data":{"data":["0930 2000.00 925","0931 1981.01 1321","0932 1984.88 1754","0933 1980.03 2033","0934 1988.98 2243","0935 1979.03 2694","0936 1981.01 3102","0937 1984.50 3430","0938 1982.97 3542","0939 1987.69 3709","0940 1991.88 3959","0941 1994.94 4240","0942 2000.01 4499","0943 2003.02 4781","0944 2002.06 5038","0945 1999.98 5260","0946 1999.01 5390","0947 1999.00 5469","0948 1999.86 5683","0949 2000.00 5987","0950 2008.00 6186","0951 2014.99 6487","0952 2013.90 6643","0953 2014.00 6830","0954 2012.00 6985","0955 2014.81 7153","0956 2018.99 7374","0957 2022.00 7666","0958 2024.25 8011","0959 2032.00 8418"],"date":"20210317"}}}}"""
if analyzer.load_from_json(sample_json):
if analyzer.calculate_averages():
analyzer.plot_chart()
更多推荐
6
0- 0
已为社区贡献22条内容
所有评论(0)