加密货币量化交易领域，真实可靠的历史逐笔成交数据是回测系统的核心资产。OKX作为头部交易所，其API提供了丰富的市场数据结构，但官方接口的限流策略、数据格式的异构性、以及批量获取时的会话管理，常常让开发者耗费大量时间在基础设施搭建而非策略研发上。
本文将系统性地梳理：如何通过OKX公开API批量获取Swap合约的历史逐笔成交数据，如何设计高效的数据清洗管线，以及在多交易所数据整合场景下，如何通过HolySheep AI的统一网关简化多源数据接入的复杂度。
为什么需要批量获取OKX逐笔成交数据
逐笔成交数据（Trade Tick Data）记录了市场上每一笔成交的精确信息：成交价格、成交量、成交方向（买入/卖出主动性）、成交时间戳。相较于K线数据，逐笔数据保留了市场的完整微观结构，是以下场景不可或缺的基础数据：
- 订单簿重构与流动性分析
- 市场冲击模型与滑点估算
- 高频策略的tick级回测
- 成交量分布与大口成交识别
- 交易所间价差套利策略的历史验证
OKX的Swap合约逐笔数据可通过RESTful API公开获取，无需认证即可访问，这对于需要大量历史回测数据的量化团队而言，是极具成本效益的数据源。
OKX逐笔成交数据API接口详解
核心接口规格
OKX提供的历史逐笔成交数据接口为 GET /api/v5/market/history-trades，支持指定合约和起始ID进行分页拉取。该接口的响应数据结构包含了进行市场微观结构分析所需的全部字段。
# OKX历史逐笔成交数据接口请求示例
Python实现，支持自动分页和错误重试
import requests
import time
import json
from datetime import datetime
class OKXTradeDataDownloader:
"""
OKX Swap合约历史逐笔成交数据批量下载器
支持自动分页、限流处理、增量同步
"""
BASE_URL = "https://www.okx.com"
def __init__(self, instrument_id: str, limit: int = 100):
"""
初始化下载器
Args:
instrument_id: OKX合约标识符，格式如 BTC-USDT-SWAP
limit: 单次请求的最大条数，最大100
"""
self.instrument_id = instrument_id
self.limit = min(limit, 100) # OKX限制单次最多100条
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'HolySheep-TradingBot/1.0'
})
def fetch_trades(self, after_id: str = None) -> dict:
"""
获取逐笔成交数据
Args:
after_id: 返回该id之后的数据，用于分页
Returns:
API响应的JSON数据
"""
endpoint = f"{self.BASE_URL}/api/v5/market/history-trades"
params = {
'instId': self.instrument_id,
'limit': self.limit
}
if after_id:
params['after'] = after_id
try:
response = self.session.get(endpoint, params=params, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
raise
def download_batch(self,
target_count: int = 10000,
target_date: str = None,
delay: float = 0.2) -> list:
"""
批量下载逐笔成交数据
Args:
target_count: 目标下载条数
target_date: 目标日期 (YYYY-MM-DD)，用于增量同步
delay: 请求间隔(秒)，避免触发限流
Returns:
包含所有成交记录的列表
"""
all_trades = []
after_id = None
consecutive_empty = 0
max_consecutive_empty = 5
print(f"开始下载 {self.instrument_id} 逐笔成交数据...")
print(f"目标条数: {target_count}, 请求间隔: {delay}s")
while len(all_trades) < target_count:
try:
data = self.fetch_trades(after_id)
if data.get('code') != '0':
error_msg = data.get('msg', 'Unknown error')
print(f"API错误: {error_msg}")
if 'rate' in error_msg.lower():
time.sleep(5) # 限流时延长等待
continue
break
trades = data.get('data', [])
if not trades:
consecutive_empty += 1
if consecutive_empty >= max_consecutive_empty:
print("连续多次返回空数据，可能已到达数据边界")
break
continue
consecutive_empty = 0
all_trades.extend(trades)
# 获取下一页的after参数
after_id = trades[-1].get('tradeId')
# 进度输出
if len(all_trades) % 1000 == 0:
print(f"已下载: {len(all_trades)} 条")
# 请求间隔控制
time.sleep(delay)
except Exception as e:
print(f"处理异常: {e}, 10秒后重试...")
time.sleep(10)
print(f"下载完成，共 {len(all_trades)} 条记录")
return all_trades
使用示例
if __name__ == "__main__":
downloader = OKXTradeDataDownloader(
instrument_id="BTC-USDT-SWAP",
limit=100
)
# 下载最近10000条逐笔数据
trades = downloader.download_batch(
target_count=10000,
delay=0.2
)
# 保存为JSON
with open('btc_usdt_swaps_trades.json', 'w') as f:
json.dump(trades, f, indent=2)
逐笔数据字段说明
OKX返回的逐笔成交数据包含以下关键字段，理解这些字段的含义对于后续的数据清洗至关重要：
| 字段名 | 数据类型 | 说明 | 量化策略中的应用 |
|---|---|---|---|
| instId | String | 合约ID，如 BTC-USDT-SWAP | 多合约数据合并时的标识 |
| tradeId | String | 成交ID，全局唯一递增 | 去重、分页定位、增量同步 |
| px | String | 成交价格 | 价格序列构建、波动率计算 |
| sz | String | 成交数量 | 成交量分析、大口成交识别 |
| side | String | 成交方向: buy(主动性买入) / sell(主动性卖出) | 订单流分析、买压卖压计算 |
| ts | String | 成交时间戳(毫秒) | 时间序列排序、事件对齐 |
数据清洗流程设计
原始API数据往往存在格式不一致、缺失值、异常值等问题，需要经过系统性的清洗才能用于量化策略回测。以下是一套完整的数据清洗管线设计：
# OKX逐笔成交数据清洗模块
实现数据验证、格式转换、异常处理、标准化输出
import pandas as pd
import numpy as np
from datetime import datetime, timezone
from typing import List, Dict, Optional, Tuple
import json
import hashlib
class OKXTradeDataCleaner:
"""
OKX逐笔成交数据清洗器
实现数据验证、格式标准化、异常值处理
"""
def __init__(self, instrument_id: str):
self.instrument_id = instrument_id
self.validation_stats = {
'total_records': 0,
'valid_records': 0,
'duplicate_records': 0,
'invalid_price': 0,
'invalid_volume': 0,
'outlier_records': 0,
'time_gap_records': 0
}
def load_from_json(self, filepath: str) -> List[Dict]:
"""从JSON文件加载原始数据"""
with open(filepath, 'r') as f:
return json.load(f)
def parse_timestamp(self, ts_str: str) -> datetime:
"""解析OKX毫秒时间戳为datetime对象"""
ts_ms = int(ts_str)
return datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
def validate_record(self, record: Dict) -> Tuple[bool, str]:
"""
验证单条记录的合法性
Returns:
(is_valid, error_message)
"""
# 检查必填字段
required_fields = ['tradeId', 'px', 'sz', 'side', 'ts']
for field in required_fields:
if field not in record or record[field] is None:
return False, f"Missing field: {field}"
# 价格验证：必须为正数
try:
price = float(record['px'])
if price <= 0:
return False, f"Invalid price: {price}"
except (ValueError, TypeError):
return False, f"Price not numeric: {record['px']}"
# 成交量验证：必须为正数
try:
volume = float(record['sz'])
if volume <= 0:
return False, f"Invalid volume: {volume}"
except (ValueError, TypeError):
return False, f"Volume not numeric: {record['sz']}"
# 方向验证
if record['side'] not in ['buy', 'sell']:
return False, f"Invalid side: {record['side']}"
return True, ""
def detect_price_outliers(self, df: pd.DataFrame,
z_threshold: float = 5.0) -> pd.Series:
"""
基于Z-Score检测价格异常值
Args:
df: 包含price列的DataFrame
z_threshold: Z分数阈值，超过此值判定为异常
Returns:
布尔Series，True表示异常
"""
mean_price = df['price'].mean()
std_price = df['price'].std()
if std_price == 0:
return pd.Series([False] * len(df), index=df.index)
z_scores = np.abs((df['price'] - mean_price) / std_price)
return z_scores > z_threshold
def detect_time_gaps(self, df: pd.DataFrame,
max_gap_seconds: int = 60) -> pd.Series:
"""
检测时间序列中的大间隔
Args:
df: 包含timestamp列的DataFrame
max_gap_seconds: 最大允许间隔(秒)
Returns:
布尔Series，True表示存在大间隔
"""
time_diffs = df['timestamp'].diff().dt.total_seconds()
return time_diffs > max_gap_seconds
def clean(self, raw_trades: List[Dict]) -> pd.DataFrame:
"""
执行完整的数据清洗流程
Args:
raw_trades: 原始OKX API数据列表
Returns:
清洗后的DataFrame
"""
self.validation_stats['total_records'] = len(raw_trades)
cleaned_records = []
seen_ids = set()
for record in raw_trades:
# 去重检查
trade_id = record.get('tradeId')
if trade_id in seen_ids:
self.validation_stats['duplicate_records'] += 1
continue
seen_ids.add(trade_id)
# 记录验证
is_valid, error_msg = self.validate_record(record)
if not is_valid:
if 'price' in error_msg.lower():
self.validation_stats['invalid_price'] += 1
else:
self.validation_stats['invalid_volume'] += 1
continue
try:
clean_record = {
'trade_id': trade_id,
'instrument_id': record.get('instId', self.instrument_id),
'price': float(record['px']),
'volume': float(record['sz']),
'side': record['side'],
'timestamp': self.parse_timestamp(record['ts']),
'trade_value': float(record['px']) * float(record['sz'])
}
cleaned_records.append(clean_record)
self.validation_stats['valid_records'] += 1
except Exception as e:
print(f"解析异常: {e}, 记录: {record}")
continue
# 创建DataFrame
if not cleaned_records:
return pd.DataFrame()
df = pd.DataFrame(cleaned_records)
# 时间排序
df = df.sort_values('timestamp').reset_index(drop=True)
# 异常值检测
outliers = self.detect_price_outliers(df)
self.validation_stats['outlier_records'] = outliers.sum()
# 时间间隔检测
time_gaps = self.detect_time_gaps(df)
self.validation_stats['time_gap_records'] = time_gaps.sum()
return df
def add_features(self, df: pd.DataFrame,
window_sizes: List[int] = [10, 50, 100]) -> pd.DataFrame:
"""
在清洗后的数据基础上添加技术特征
Args:
df: 清洗后的DataFrame
window_sizes: 计算滚动窗口的窗口大小列表
Returns:
添加特征后的DataFrame
"""
if df.empty:
return df
# 添加收益率
df['return'] = df['price'].pct_change()
# 添加成交量滚动统计
for window in window_sizes:
df[f'volume_ma_{window}'] = df['volume'].rolling(window=window).mean()
df[f'trade_count_{window}'] = df['volume'].rolling(window=window).count()
# 添加主动性买卖标识数值化
df['side_numeric'] = df['side'].map({'buy': 1, 'sell': -1})
# 添加买卖不平衡指标
for window in window_sizes:
df[f'order_imbalance_{window}'] = (
df['side_numeric'].rolling(window=window).sum() / window
)
# 添加价格波动率
for window in window_sizes:
df[f'volatility_{window}'] = df['return'].rolling(window=window).std() * np.sqrt(window)
return df
def get_validation_report(self) -> Dict:
"""生成数据质量报告"""
total = self.validation_stats['total_records']
valid = self.validation_stats['valid_records']
return {
'summary': {
'total_records': total,
'valid_records': valid,
'valid_rate': f"{valid/total*100:.2f}%" if total > 0 else "N/A"
},
'issue_breakdown': {
'duplicates': self.validation_stats['duplicate_records'],
'invalid_price': self.validation_stats['invalid_price'],
'invalid_volume': self.validation_stats['invalid_volume'],
'price_outliers': self.validation_stats['outlier_records'],
'time_gaps': self.validation_stats['time_gap_records']
}
}
使用示例
if __name__ == "__main__":
# 初始化清洗器
cleaner = OKXTradeDataCleaner(instrument_id="BTC-USDT-SWAP")
# 加载原始数据
raw_data = cleaner.load_from_json('btc_usdt_swaps_trades.json')
# 执行清洗
df_clean = cleaner.clean(raw_data)
# 添加技术特征
df_features = cleaner.add_features(df_clean)
# 输出质量报告
report = cleaner.get_validation_report()
print("\n=== 数据质量报告 ===")
print(f"总记录数: {report['summary']['total_records']}")
print(f"有效记录: {report['summary']['valid_records']}")
print(f"有效率: {report['summary']['valid_rate']}")
print(f"重复记录: {report['issue_breakdown']['duplicates']}")
print(f"价格异常: {report['issue_breakdown']['price_outliers']}")
# 保存清洗结果
df_features.to_parquet('btc_usdt_trades_cleaned.parquet', index=False)
print("\n清洗完成，数据已保存为Parquet格式")
HolySheep AI vs 官方API vs 其他方案
在量化交易系统的数据获取层面，开发者通常面临多种技术路线的选择。以下从关键维度进行系统对比：
| 对比维度 | HolySheep AI | OKX官方API | CCXT开源库 | 付费数据商 |
|---|---|---|---|---|
| 数据覆盖 | 多交易所聚合 | 仅OKX单一交易所 | 40+交易所 | 通常3-10家 |
| 请求限制 | 统一限流管理 | 严格限流，需自行实现重试 | 各交易所限制不同 | 通常无严格限制 |
| 数据完整性 | 自动去重、校验 | 需自行清洗 | 基础标准化 | 专业清洗 |
| 历史数据深度 | 视合作深度 | 有限（部分品种） | 依赖交易所 | 多年历史 |
| 集成复杂度 | 统一SDK | 原生API | 中等 | API/文件 |
| 使用成本 | 订阅制 | 免费（限流） | 免费开源 | 按量计费 |
| 技术支持 | 专业支持 | 社区文档 | 社区支持 | 企业级支持 |
这种方案适合 / 不适合的场景
适合使用OKX原生API获取逐笔数据的场景
- 单一交易所专注策略：策略仅针对OKX合约，不需要多交易所数据
- 成本敏感型项目：预算有限，能够接受API限流和自行维护数据管线
- 短期回测需求：仅需近期数据，API免费额度足够
- 技术能力充足团队：有专职工程师能够处理API稳定性、错误重试、数据清洗
不适合使用单一交易所API的场景
- 跨交易所套利策略：需要实时获取多家交易所数据进行价差计算
- 深度历史回测：需要数年的完整tick数据进行策略验证
- 高频交易系统：对数据延迟和完整性有严格要求
- 研发效率优先：希望将工程资源集中在策略研发而非数据基础设施
数据获取的进阶架构设计
对于需要长期运行的生产级量化系统，建议采用以下架构设计，确保数据的可靠性和系统的稳定性：
# 生产级逐笔数据采集与存储架构
实现增量同步、分布式存储、数据订阅
import asyncio
import aiohttp
import redis
import pandas as pd
from typing import Optional, Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class SyncState:
"""同步状态记录"""
instrument_id: str
last_trade_id: str
last_timestamp: datetime
sync_count: int
last_sync_time: datetime
class ProductionTradeCollector:
"""
生产级逐笔成交数据采集器
特性：
- 增量同步，只获取新数据
- Redis状态持久化，支持服务重启恢复
- 异步并发请求
- 数据管道化处理
"""
def __init__(self,
instrument_ids: list,
redis_client: redis.Redis,
storage_callback: Callable[[pd.DataFrame], None]):
self.instrument_ids = instrument_ids
self.redis = redis_client
self.storage = storage_callback
self.base_url = "https://www.okx.com/api/v5/market/history-trades"
self.semaphore = asyncio.Semaphore(3) # 限制并发数
def _get_state_key(self, instrument_id: str) -> str:
return f"okx:trade:sync:{instrument_id}"
def get_sync_state(self, instrument_id: str) -> Optional[SyncState]:
"""从Redis获取同步状态"""
key = self._get_state_key(instrument_id)
state_json = self.redis.get(key)
if not state_json:
return None
import json
state_dict = json.loads(state_json)
return SyncState(
instrument_id=state_dict['instrument_id'],
last_trade_id=state_dict['last_trade_id'],
last_timestamp=datetime.fromisoformat(state_dict['last_timestamp']),
sync_count=state_dict['sync_count'],
last_sync_time=datetime.fromisoformat(state_dict['last_sync_time'])
)
def update_sync_state(self, state: SyncState):
"""更新Redis中的同步状态"""
key = self._get_state_key(state.instrument_id)
import json
state_dict = {
'instrument_id': state.instrument_id,
'last_trade_id': state.last_trade_id,
'last_timestamp': state.last_timestamp.isoformat(),
'sync_count': state.sync_count,
'last_sync_time': datetime.now(timezone.utc).isoformat()
}
self.redis.setex(key, 86400 * 7, json.dumps(state_dict)) # 7天过期
async def fetch_trades_async(self,
session: aiohttp.ClientSession,
instrument_id: str,
after_id: Optional[str] = None) -> dict:
"""异步获取单次请求"""
async with self.semaphore:
params = {'instId': instrument_id, 'limit': 100}
if after_id:
params['after'] = after_id
try:
async with session.get(self.base_url, params=params) as resp:
if resp.status == 429:
logger.warning(f"限流触发，等待重试: {instrument_id}")
await asyncio.sleep(5)
return await self.fetch_trades_async(session, instrument_id, after_id)
data = await resp.json()
return data.get('data', [])
except Exception as e:
logger.error(f"请求异常: {instrument_id}, {e}")
return []
async def sync_single_instrument(self, instrument_id: str) -> int:
"""同步单个合约的增量数据"""
state = self.get_sync_state(instrument_id)
after_id = state.last_trade_id if state else None
all_trades = []
consecutive_empty = 0
max_iterations = 1000 # 防止无限循环
async with aiohttp.ClientSession() as session:
for _ in range(max_iterations):
trades = await self.fetch_trades_async(session, instrument_id, after_id)
if not trades:
consecutive_empty += 1
if consecutive_empty >= 3:
break
await asyncio.sleep(0.2)
continue
consecutive_empty = 0
all_trades.extend(trades)
after_id = trades[-1]['tradeId']
# 每1000条处理一次
if len(all_trades) >= 1000:
df = self._process_trades(all_trades)
self.storage(df)
all_trades = []
await asyncio.sleep(0.2) # 避免触发限流
# 处理剩余数据
if all_trades:
df = self._process_trades(all_trades)
self.storage(df)
return len(all_trades)
def _process_trades(self, trades: list) -> pd.DataFrame:
"""将原始记录转换为DataFrame"""
records = []
for t in trades:
records.append({
'trade_id': t['tradeId'],
'instrument_id': t.get('instId'),
'price': float(t['px']),
'volume': float(t['sz']),
'side': t['side'],
'timestamp': datetime.fromtimestamp(int(t['ts'])/1000, tz=timezone.utc),
'collected_at': datetime.now(timezone.utc)
})
return pd.DataFrame(records)
async def run(self, interval_seconds: int = 60):
"""启动定时同步任务"""
while True:
tasks = [
self.sync_single_instrument(inst_id)
for inst_id in self.instrument_ids
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for inst_id, result in zip(self.instrument_ids, results):
if isinstance(result, Exception):
logger.error(f"同步失败: {inst_id}, {result}")
else:
logger.info(f"同步完成: {inst_id}, 新增 {result} 条")
await asyncio.sleep(interval_seconds)
存储回调示例：写入Parquet分区表
def storage_to_parquet(df: pd.DataFrame):
"""将数据追加写入Parquet分区表"""
if df.empty:
return
partition_path = f"data/trades/year={df['timestamp'].dt.year}/month={df['timestamp'].dt.month:02d}/day={df['timestamp'].dt.day:02d}/trades.parquet"
df.to_parquet(partition_path, partition_cols=['timestamp'], append=True)
使用示例
if __name__ == "__main__":
redis_client = redis.Redis(host='localhost', port=6379, db=0)
collector = ProductionTradeCollector(
instrument_ids=[
"BTC-USDT-SWAP",
"ETH-USDT-SWAP",
"SOL-USDT-SWAP"
],
redis_client=redis_client,
storage_callback=storage_to_parquet
)
# 运行同步任务
asyncio.run(collector.run(interval_seconds=60))
HolySheep AI注册と бесплатные кредиты
HolySheep AI作为全球领先的AI API网关服务，不仅在AI模型调用方面提供卓越的统一接入体验，其开放的生态体系也支持与各类数据源的无缝集成。对于量化研究团队而言，HolySheep AI的核心价值在于：
- 统一认证体系：单点接入，多数据源授权管理
- 成本优化：相比单独对接多个数据源，综合成本降低40-60%
- 稳定性保障：99.9%的可用性SLA，企业级服务保障
- 技术赋能：专业的量化领域技术支持团队
常见问题与解决方案
1. API返回空数据但未报错
问题描述：请求API后返回200状态码，但data数组为空，且无错误信息。
原因分析：这种情况通常发生在两种场景：请求的时间范围内确实没有新成交，或者after参数设置不当导致查询范围错误。
解决方案：
# 问题诊断与修复代码
def diagnose_empty_response(instrument_id: str, after_id: str = None):
"""
诊断API返回空数据的可能原因
"""
import requests
url = "https://www.okx.com/api/v5/market/history-trades"
params = {'instId': instrument_id, 'limit': 100}
if after_id:
params['after'] = after_id
response = requests.get(url, params=params)
data = response.json()
print(f"响应代码: {data.get('code')}")
print(f"响应消息: {data.get('msg')}")
print(f"数据条数: {len(data.get('data', []))}")
# 如果有数据，检查最新记录的ID和时间
if data.get('data'):
latest = data['data'][0]
print(f"最新成交ID: {latest.get('tradeId')}")
print(f"最新成交时间: {latest.get('ts')}")
# 如果after_id之后没有数据，尝试向前查询
if after_id and latest.get('tradeId') == after_id:
print("警告: after参数之后没有新数据，可能是数据已同步完成")
print("建议: 检查本地最后同步ID是否正确")
return data
使用示例
result = diagnose_empty_response("BTC-USDT-SWAP")
2. 请求触发限流(429错误)
问题描述：高频请求后收到HTTP 429错误，API返回限流提示。
原因分析：OKX API对公开市场数据的默认限流为每秒20次请求。超出限制会触发限流保护。
解决方案：
# 指数退避重试机制实现
import time
import random
from functools import wraps
def exponential_backoff_retry(max_retries: int = 5, base_delay: float = 1.0):
"""
指数退避重试装饰器
适用于API限流场景
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if '429' in str(e) or 'rate' in str(e).lower():
# 计算退避时间：base_delay * 2^attempt + 随机抖动
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"限流触发，等待 {delay:.2f}秒后重试 (第{attempt+1}次)")
time.sleep(delay)
else:
raise
raise Exception(f"超过最大重试次数 {max_retries}")
return wrapper
return decorator
使用示例
@exponential_backoff_retry(max_retries=5, base_delay=2.0)
def fetch_with_retry(url: str, params: dict):
response = requests.get(url, params=params, timeout=30)
if response.status_code == 429:
raise Exception("429: Rate limit exceeded")
return response.json()
3. 数据中存在价格异常跳变
问题描述：清洗后的数据中发现个别成交价格与周围数据存在显著偏离，可能是异常成交或数据错误。
原因分析：交易所系统故障、插针现象、或数据传输过程中的错误都可能导致异常价格记录。
解决方案：
# 多维度异常值检测与处理
def advanced_outlier_detection(df: pd.DataFrame,
price_z_threshold: float = 4.0,
volume_z_threshold: float = 6.0,
price_change_threshold: float = 0.05) -> pd.DataFrame:
"""
多维度异常值检测
检测维度：
1. 绝对价格Z-Score异常
2. 成交量Z-Score异常
3. 相对价格跳变异常
"""
df = df.copy()
# 维度1: 价格Z-Score
price_mean = df['price'].mean()
price_std = df['price'].std()
df['price_zscore'] = abs((df['price'] - price_mean) / price_std) if price_std > 0 else 0
df['is_price_outlier'] = df['price_zscore'] > price_z_threshold
# 维度2: 成交量Z-Score
volume_mean = df['volume'].mean()
volume_std = df['volume'].std()
df['volume_zscore'] = abs((df['volume'] - volume_mean) / volume_std) if volume_std > 0 else 0
df['is_volume_outlier'] = df['volume_zscore'] > volume_z_threshold
# 维度3: 价格跳变检测
df['price_change'] = df['price'].pct_change().abs()
df['is_price_jump'] = df['price_change'] > price_change_threshold
# 综合异常标记
df['is_anomaly'] = (
df['is_price_outlier'] |
df['is_volume_outlier'] |
df['is_price_jump']
)
# 统计异常数量
anomaly_count = df['is_anomaly'].sum()
print(f"检测到 {anomaly_count} 条异常记录 ({anomaly_count/len(df