你在对接股票数据接口API时，是否遇到过实时行情推送看似正常，但历史数据里偶尔出现时间倒序、价格跳变或成交量为空的情况？这类异常如果不处理，会直接影响 K 线生成和指标计算。本文提供一套轻量级的过滤方案，包含时间顺序、价格波动和字段完整性校验。
问题背景
实时行情属于连续变化的数据流，程序接收和处理过程中可能出现以下异常：
- 时间戳顺序异常，新数据时间早于上一条记录；
- 价格变化幅度明显超过正常范围；
- 成交量字段缺失或格式错误；
- 股票代码、交易状态等关键字段为空。
这些数据如果直接写入数据库，后续生成 K 线时容易出现偏差。例如一分钟 K 线中混入错误价格，最高价、最低价计算失真，进一步影响技术指标分析。
时间顺序校验
处理实时行情时，优先检查时间字段。正常情况下，同一股票的推送时间应该保持递增。如果收到一条时间更早的数据，暂时过滤，避免影响后续排列。简单判断逻辑如下：
def check_time(last_time, current_time):
if current_time < last_time:
return False
return True
实际开发中，时间校验还需要结合交易所时区、服务器时间以及数据源时间。美股行情在夏令时切换期间尤其需要注意时间转换。
价格波动校验
价格校验是另一个重要环节。股票价格不会在极短时间内无限制变化，可以根据历史波动范围设置判断条件。例如：
def check_price(old_price, new_price):
change = abs(new_price - old_price) / old_price
if change > 0.1:
return False
return True
阈值需要根据不同股票类型调整。大型股票和波动较高的股票，判断标准不能完全一样。
字段完整性校验
除了价格和时间，还需要检查行情数据结构是否完整。收到一条推送数据后，不要直接进入计算模块，而是先确认：
symbol是否存在；price是否为空；volume格式是否正确；timestamp是否有效。
例如使用 AllTick API 的 WebSocket 获取股票行情时，可以先完成字段检查，再把数据交给后续处理。
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
if not data.get("symbol"):
return
if not data.get("price"):
return
print(data["symbol"], data["price"])
ws = websocket.WebSocketApp(
"wss://apis.alltick.co/websocket-api/stock-websocket-interface-api/transaction-quote-subscription",
on_message=on_message
)
ws.run_forever()
这样可以避免错误数据直接影响行情计算。
异常记录方案
直接过滤异常数据虽然简单，但更推荐保留拦截记录。每条被拦截的数据可以保存：
- 股票代码；
- 数据时间；
- 原始价格；
- 异常原因。
后续查看这些记录，可以判断是数据源波动，还是程序逻辑需要调整。
总结
做股票行情系统之后，数据质量会越来越重要。接口连接只是第一步，真正影响分析结果的是进入系统里的每一条数据是否可靠。股票数据接口API提供的是行情来源，而数据校验负责保证这些数据能够被正确使用。增加简单的数据检查机制，往往能减少很多隐藏风险。
