2021 年 3 月，Bill Hwang 的家族办公室 Archegos Capital Management 在两天内亏损超过 100 亿美元，引发了人类历史上最大的个股单日跌幅之一。事后调查显示，Archegos 的交易日志极不完整——没有清晰的仓位归因、没有完整的衍生品估值链条、没有可追溯至每一笔决策的交易链路。当 SEC 和 FINRA 的调查人员要求还原 2020 年 3 月某日的风险敞口时，团队花了三周时间也无法给出一个令监管信服的数字。
这不是极端案例。2022 年以来，全球主要监管机构对量化交易的审查力度显著加强。SEC 修订了 Rule 15c3-5（市场准入规则），要求所有算法交易商保留完整的订单生命周期记录；MiFID II 的交易重构要求（Transaction Reconstruction）规定，所有交易活动必须能够精确还原到毫秒级时间戳；中国的《证券期货业数据分类分级指引》则对期货公司、量化私募提出了明确的日志留存要求。
合规审计不再是“出了问题再补”的事后动作，而是交易系统设计的原生需求。
本文从监管要求与内部风控两个维度，系统拆解量化交易场景下的日志体系设计，提供可落地的数据留存策略，并给出生产级的日志采集与归档代码。
一、为什么合规审计是量化系统的“基础设施”
量化交易的合规审计面临两个核心驱动力的拉扯：外部监管要求与内部风控需求。二者关注的日志类型有重叠，但在完整性和时效性上存在显著差异。
1.1 监管要求的日志类型
| 监管框架 | 关键要求 | 时间跨度 | 粒度要求 |
|---|---|---|---|
| SEC Rule 15c3-5 | 订单录入、修改、取消、成交全链路 | 6 年 | 毫秒级时间戳 |
| MiFID II Article 25 | 交易重构能力，订单与成交关联 | 5 年 | 微秒级时间戳 |
| CFTC 规则 17 CFR 1.31 | 电子记录保留（审计追踪） | 5 年 | 秒级，需防篡改 |
| 中国《证券期货经营机构私募资产管理业务管理办法》 | 投资决策、交易执行、清算交收 | 20 年 | 秒级，纸质或电子 |
监管要求的核心特征是：留痕 + 不可篡改 + 可还原。审计人员不需要实时访问这些数据，但当调查启动时，必须能够在可接受的时间窗口内（如 48 小时）提供完整的交易重构报告。
1.2 内部风控的日志需求
内部风控的关注点与监管不同——它更强调实时性和可操作性。
| 风控场景 | 需要的日志类型 | 典型时效要求 |
|---|---|---|
| 策略回撤监控 | 每日PnL、持仓快照、因子暴露 | 实时或 T+1 |
| 异常交易检测 | 订单频率、撤销率、价格偏离度 | 分钟级告警 |
| 系统故障溯源 | 网络延迟、API错误码、连接重置 | 秒级定位 |
| 策略失效复盘 | 信号生成、订单提交、执行回报的完整链路 | T+N 查询 |
内部风控的日志留存周期通常短于监管要求（热数据 30-90 天，温数据 1-2 年），但完整性要求同样严格——因为一次策略失效的复盘，可能需要追溯到半年前的市场状态。
1.3 两套需求的交汇点
无论监管还是内部风控，以下日志类型是必须留存的核心资产：
- 交易执行日志：订单生命周期完整记录
- 市场数据日志：接收到的行情数据，用于回放与验证
- 系统运行日志：连接状态、心跳、错误码
- 策略决策日志：信号生成、参数变更、模型版本
接下来的章节将逐类展开具体的数据结构设计和技术实现方案。
二、核心日志体系设计
2.1 交易执行日志
交易执行日志是合规审计的核心资产，也是监管机构首先调取的数据类型。一个结构完整的交易执行日志需要记录订单从“想法”到“成交”的完整生命周期：
{
"log_type": "execution",
"log_id": "exec-20240601-08927341-a7f3",
"timestamp": "2024-06-01T09:30:15.847392",
"timezone": "America/New_York",
"strategy_id": "strat-001",
"strategy_version": "v2.3.1",
"order": {
"client_order_id": "clord-20240601-001",
"exchange_order_id": "NYSE-8847291",
"symbol": "AAPL.US",
"side": "BUY",
"order_type": "LIMIT",
"tif": "DAY",
"price": 185.50,
"quantity": 1000,
"filled_quantity": 0,
"avg_fill_price": null,
"status": "NEW",
"created_at": "2024-06-01T09:30:15.847392",
"updated_at": "2024-06-01T09:30:15.847392"
},
"events": [
{
"event_type": "ORDER_SUBMITTED",
"timestamp": "2024-06-01T09:30:15.847392",
"latency_ns": null,
"details": {}
},
{
"event_type": "ORDER_SENT_TO_EXCHANGE",
"timestamp": "2024-06-01T09:30:15.849201",
"latency_ns": 1809,
"details": {
"destination": "ARCA",
"transmission_time_ns": 1205
}
},
{
"event_type": "ORDER_ACKNOWLEDGED",
"timestamp": "2024-06-01T09:30:15.862847",
"latency_ns": 15455,
"details": {
"exchange_timestamp": "2024-06-01T09:30:15.861000"
}
}
],
"source_ip": "10.0.1.45",
"host_id": "trader-node-03"
}
设计要点解读：
- 双重时间戳：订单的
created_at（系统时间）和交易所回传的exchange_timestamp必须同时记录，用于计算网络延迟和检测时间同步问题。 - 事件链（Events Array）：每一笔订单的完整状态变迁必须作为不可变的追加日志存在，而不是覆盖更新。这不仅满足监管的“审计追踪”要求，也便于回放订单簿变化。
- 策略版本锁定：每个订单必须关联
strategy_version。当策略更新后出现异常时，能够快速定位是策略逻辑问题还是执行问题。 - 延迟追踪：关键节点（如从提交到交易所确认）的耗时必须以纳秒级精度记录，这对于低延迟策略的性能分析至关重要。
2.2 市场数据日志
市场数据日志的核心价值在于可回放性。当策略在某个交易日出现异常表现时，需要用与当时完全一致的市场数据来回放策略逻辑，验证异常是市场因素还是策略缺陷。
{
"log_type": "market_data",
"log_id": "md-20240601-093015-00008421",
"timestamp": "2024-06-01T09:30:15.001847",
"source": "tickdb",
"source_id": "stream-connection-041",
"symbol": "AAPL.US",
"data_type": "quote",
"fields": {
"bid_price": 185.48,
"bid_size": 400,
"ask_price": 185.50,
"ask_size": 600,
"last_price": 185.49,
"last_size": 100,
"volume": 1245000,
"quote_condition": "NORMAL"
},
"metadata": {
"feed_latency_us": 87,
"local_receive_time": "2024-06-01T09:30:15.001760",
"feed_source": "CTA"
},
"checksum": "sha256:a3f8c2d..."
}
设计要点解读：
- 来源标注：
source字段明确记录数据来源。当使用多数据源聚合时，这个字段对于排查“哪个数据源出现了延迟或错误”至关重要。 - 延迟度量：
feed_latency_us记录了数据从源站到本地的时间差，是低延迟策略风控的核心指标。 - 校验和：每条市场数据记录都附带 SHA-256 校验和，用于验证数据完整性——特别是当需要用这些数据做合规回放时，必须确保数据未被篡改。
2.3 策略决策日志
策略决策日志记录的是“策略在想什么”，是连接信号生成与交易执行的桥梁。
{
"log_type": "strategy_decision",
"log_id": "decision-20240601-093015-00000123",
"timestamp": "2024-06-01T09:30:15.001",
"strategy_id": "strat-001",
"strategy_version": "v2.3.1",
"signal": {
"type": "MOMENTUM_LONG",
"symbol": "AAPL.US",
"direction": 1,
"strength": 0.78,
"factors": {
"price_momentum_20d": 0.45,
"volume_surge_5m": 1.23,
"spread_compression": 0.32
}
},
"context": {
"portfolio_leverage": 1.8,
"sector_exposure": {
"Technology": 0.42,
"Consumer": 0.15
},
"var_95_1d": 125000
},
"decision": {
"action": "SUBMIT_ORDER",
"quantity": 1000,
"order_type": "LIMIT",
"price": 185.50,
"rationale": "Signal strength 0.78 exceeds threshold 0.6, within risk limits"
}
}
设计要点解读：
- 因子可解释性：每个信号都附带因子拆解（
factors），不仅满足监管的“算法决策透明性”要求，也便于风控团队验证策略是否按预期运行。 - 上下文快照：记录决策时刻的组合状态（杠杆、行业敞口、VaR），用于事后评估“当时的决策是否在风险边界内”。
- 决策链路：从信号到订单的推导过程必须完整记录。当审计人员问“为什么在这个价格买了 1000 股”时，能够给出量化依据。
三、日志归档架构设计
3.1 分层存储策略
不同的日志类型有不同的访问模式，必须采用分层存储策略：
| 存储层 | 介质 | 保留周期 | 适用场景 | 典型工具 |
|---|---|---|---|---|
| 热数据层 | SSD NVMe / 内存 | 0-7 天 | 实时监控、异常告警 | Elasticsearch、ClickHouse |
| 温数据层 | HDD SAN / 本地 SSD | 7-90 天 | 日常复盘、临时查询 | Elasticsearch、S3 + 索引 |
| 冷数据层 | 对象存储 | 90 天 - 合规要求 | 合规审计、监管调取 | S3、Azure Blob、Google Cloud Storage |
设计原则：
- 热数据层优先考虑写入性能和查询延迟，适合毫秒级监控告警场景
- 温数据层需要在存储成本和查询速度之间取得平衡，适合周级别的策略复盘
- 冷数据层以成本优先，但必须支持按需恢复——当监管调取时，能够在 48 小时内提供完整数据
3.2 架构总览
┌─────────────────────────────────────────────────────────────────────────┐
│ 日志归档架构 │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 交易执行 │ │ 市场数据 │ │ 策略决策 │ │
│ │ 日志生成 │ │ 日志生成 │ │ 日志生成 │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ └────────────────────────┼────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ Kafka │ 消息队列：解耦、缓冲、 │
│ │ Cluster │ 顺序保证、重放能力 │
│ └──────┬───────┘ │
│ │ │
│ ┌────────────────────┼────────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Logstash / │ │ Logstash / │ │ Logstash / │ │
│ │ Vector │ │ Vector │ │ Vector │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Elasticsearch│ │ Elasticsearch│ │ S3 Sink │ │
│ │ (热/温层) │ │ (热/温层) │ │ (冷数据) │ │
│ └──────────────┘ └──────────────┘ └──────┬───────┘ │
│ │ │ │ │
│ │ │ ▼ │
│ │ │ ┌──────────────┐ │
│ │ │ │ Glacier │ │
│ │ │ │ (超冷存储) │ │
│ │ │ └──────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Kibana / │ │ 合规查询 │ │
│ │ Grafana │ │ 报告生成 │ │
│ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
架构说明：
- Kafka 作为统一入口：所有日志类型先写入 Kafka，实现解耦——日志生成方不需要关心下游存储策略，Kafka 的持久化特性也提供了天然的数据备份。
- 双写策略：Logstash/Vector 同时写入 Elasticsearch（热/温层）和 S3（冷数据层），保证数据既有快速查询能力，又有合规归档保障。
- 冷热分层自动化：通过 Kafka Connect 的 S3 Sink Connector，可以设置基于时间的路由策略——7 天前的数据自动从 Elasticsearch 过期，迁移到 S3。
四、生产级日志采集与归档代码
4.1 日志采集器：Python + Kafka Producer
import os
import json
import time
import uuid
import hashlib
import logging
from datetime import datetime, timezone
from kafka import KafkaProducer
from kafka.errors import KafkaError
import threading
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class CompliantLogger:
"""
符合监管要求的交易日志采集器。
特性：
- 结构化 JSON 输出，满足审计追踪要求
- Kafka 异步写入，不阻塞交易主流程
- 自动校验和计算，支持数据完整性验证
- 指数退避重连，保证高可用
"""
def __init__(self, bootstrap_servers=None, api_key=None):
self.bootstrap_servers = bootstrap_servers or os.environ.get('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
self.api_key = api_key or os.environ.get('TICKDB_API_KEY')
self.producer = None
self._connect_lock = threading.Lock()
self._reconnect_attempts = 0
self._max_reconnect_attempts = 10
self._base_reconnect_delay = 1
self._max_reconnect_delay = 60
def _connect(self):
"""建立 Kafka 连接，带指数退避和抖动"""
with self._connect_lock:
if self.producer is not None:
return True
try:
self.producer = KafkaProducer(
bootstrap_servers=self.bootstrap_servers,
value_serializer=lambda v: json.dumps(v, default=str).encode('utf-8'),
key_serializer=lambda k: k.encode('utf-8') if k else None,
acks='all', # 等待所有副本确认，保证持久性
retries=3,
max_in_flight_requests_per_connection=1, # 保证顺序
linger_ms=5, # 批量发送，降低延迟
compression_type='gzip', # 压缩存储
# ⚠️ 生产环境建议使用 SSL/SASL 认证
)
self._reconnect_attempts = 0
logger.info(f"Kafka producer connected to {self.bootstrap_servers}")
return True
except KafkaError as e:
logger.error(f"Failed to connect to Kafka: {e}")
return self._handle_connection_failure()
def _handle_connection_failure(self):
"""连接失败处理：指数退避 + 抖动"""
self._reconnect_attempts += 1
if self._reconnect_attempts > self._max_reconnect_attempts:
logger.error("Max reconnection attempts reached, logging to local file")
return False
delay = min(
self._base_reconnect_delay * (2 ** (self._reconnect_attempts - 1)),
self._max_reconnect_delay
)
# 添加抖动，避免惊群效应
import random
jitter = random.uniform(0, delay * 0.1)
sleep_time = delay + jitter
logger.warning(f"Reconnecting in {sleep_time:.2f}s (attempt {self._reconnect_attempts})")
time.sleep(sleep_time)
return self._connect()
def _generate_log_id(self, prefix: str) -> str:
"""生成唯一的日志 ID"""
return f"{prefix}-{datetime.now(timezone.utc).strftime('%Y%m%d')}-{uuid.uuid4().hex[:12]}"
def _calculate_checksum(self, data: dict) -> str:
"""计算 SHA-256 校验和，用于数据完整性验证"""
content = json.dumps(data, sort_keys=True, default=str)
return hashlib.sha256(content.encode('utf-8')).hexdigest()
def log_execution(self, order_data: dict, events: list) -> bool:
"""
记录交易执行日志。
Args:
order_data: 订单信息字典
events: 订单状态变更事件列表
Returns:
bool: 记录是否成功
"""
log_entry = {
"log_type": "execution",
"log_id": self._generate_log_id("exec"),
"timestamp": datetime.now(timezone.utc).isoformat(),
"timezone": "America/New_York",
"strategy_id": order_data.get("strategy_id"),
"strategy_version": order_data.get("strategy_version", "unknown"),
"order": order_data,
"events": events,
"source_ip": self._get_local_ip(),
"host_id": os.environ.get('HOSTNAME', 'unknown'),
"checksum": None # 先计算再填入
}
# 计算校验和（不包含 checksum 字段本身）
log_entry["checksum"] = self._calculate_checksum(log_entry)
return self._send_to_kafka("trading-execution-logs", order_data.get("client_order_id"), log_entry)
def log_market_data(self, symbol: str, data_type: str, fields: dict, metadata: dict = None) -> bool:
"""
记录市场数据日志。
Args:
symbol: 交易品种代码
data_type: 数据类型 (quote/trade/depth)
fields: 行情数据字段
metadata: 额外元数据（如延迟、来源等）
"""
log_entry = {
"log_type": "market_data",
"log_id": self._generate_log_id("md"),
"timestamp": datetime.now(timezone.utc).isoformat(),
"source": "tickdb", # 标注数据来源，便于审计
"source_id": os.environ.get('TICKDB_CONNECTION_ID', 'unknown'),
"symbol": symbol,
"data_type": data_type,
"fields": fields,
"metadata": metadata or {},
"checksum": None
}
log_entry["checksum"] = self._calculate_checksum(log_entry)
return self._send_to_kafka("market-data-logs", symbol, log_entry)
def log_strategy_decision(self, signal: dict, context: dict, decision: dict) -> bool:
"""
记录策略决策日志。
"""
log_entry = {
"log_type": "strategy_decision",
"log_id": self._generate_log_id("decision"),
"timestamp": datetime.now(timezone.utc).isoformat(),
"strategy_id": signal.get("strategy_id", "unknown"),
"strategy_version": signal.get("strategy_version", "unknown"),
"signal": signal,
"context": context,
"decision": decision,
"checksum": None
}
log_entry["checksum"] = self._calculate_checksum(log_entry)
return self._send_to_kafka("strategy-decision-logs", signal.get("symbol"), log_entry)
def _send_to_kafka(self, topic: str, key: str, message: dict) -> bool:
"""发送消息到 Kafka，带错误处理和本地回退"""
if self._connect():
try:
future = self.producer.send(topic, key=key, value=message)
# 异步等待确认，不阻塞交易主流程
# ⚠️ 高频交易场景建议完全异步，通过回调处理失败重试
future.add_callback(self._on_send_success)
future.add_errback(self._on_send_error)
return True
except KafkaError as e:
logger.error(f"Failed to send log to Kafka: {e}")
return self._fallback_to_local_file(topic, message)
else:
return self._fallback_to_local_file(topic, message)
def _on_send_success(self, record_metadata):
"""发送成功回调"""
logger.debug(f"Log sent to {record_metadata.topic}:{record_metadata.partition}:{record_metadata.offset}")
def _on_send_error(self, exception):
"""发送失败回调：记录错误并触发重连"""
logger.error(f"Failed to send log: {exception}")
self._reconnect_attempts = self._max_reconnect_attempts # 触发重连
self._connect()
def _fallback_to_local_file(self, topic: str, message: dict) -> bool:
"""Kafka 不可用时的本地回退：保证日志不丢失"""
try:
local_path = f"/var/log/trading/{topic}-{datetime.now().strftime('%Y%m%d')}.jsonl"
with open(local_path, 'a') as f:
f.write(json.dumps(message, default=str) + '\n')
logger.warning(f"Log written to local fallback: {local_path}")
return True
except Exception as e:
logger.error(f"Failed to write to local fallback: {e}")
return False
def _get_local_ip(self) -> str:
"""获取本机 IP 地址"""
import socket
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
except Exception:
return "unknown"
def flush(self):
"""手动刷新缓冲区"""
if self.producer:
self.producer.flush()
def close(self):
"""关闭连接"""
if self.producer:
self.producer.flush()
self.producer.close()
self.producer = None
4.2 自动化归档任务：S3 冷数据迁移
import os
import json
import boto3
from botocore.config import Config
from datetime import datetime, timezone, timedelta
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LogArchiver:
"""
日志冷数据归档器。
功能：
- 将 Elasticsearch 中的历史日志归档到 S3
- 支持合规要求的长期留存
- 自动清理已归档数据，释放存储空间
"""
def __init__(self, aws_access_key=None, aws_secret_key=None, region_name='us-east-1'):
session = boto3.Session(
aws_access_key_id=aws_access_key or os.environ.get('AWS_ACCESS_KEY_ID'),
aws_secret_access_key=aws_secret_key or os.environ.get('AWS_SECRET_ACCESS_KEY'),
region_name=region_name
)
self.s3 = session.client('s3')
self.sts = session.client('sts')
self.bucket_name = os.environ.get('ARCHIVE_BUCKET', 'trading-compliance-logs')
self.compliance_retention_years = 7 # SEC Rule 17 CFR 1.31 默认要求
def archive_execution_logs(self, days_old: int = 7) -> dict:
"""
归档超过指定天数的交易执行日志。
Args:
days_old: 超过多少天的日志需要归档
Returns:
dict: 归档结果统计
"""
logger.info(f"Starting archive of execution logs older than {days_old} days")
cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_old)
archive_date = datetime.now(timezone.utc).strftime('%Y-%m-%d')
# ⚠️ 这里需要集成 Elasticsearch 查询
# 以下为示意代码，实际使用需替换为真实 ES 客户端
# from elasticsearch import Elasticsearch
# es = Elasticsearch([os.environ.get('ES_HOST')])
# query = {"query": {"range": {"timestamp": {"lt": cutoff_date.isoformat()}}}}
# results = es.search(index="trading-execution-logs-*", body=query, scroll="2m", size=10000)
# 模拟处理
archived_count = 0
failed_count = 0
total_size_bytes = 0
# 示例：处理一个批次的数据
batch = self._fetch_es_records("trading-execution-logs", cutoff_date)
if batch:
s3_key = f"execution-logs/year={archive_date[:4]}/month={archive_date[5:7]}/day={archive_date[8:10]}/execution-{archive_date}.json.gz"
try:
# 写入 S3，使用 gzip 压缩
self.s3.put_object(
Bucket=self.bucket_name,
Key=s3_key,
Body=self._compress_records(batch),
ContentType='application/gzip',
Metadata={
'archive-date': archive_date,
'record-count': str(len(batch)),
'source-index': 'trading-execution-logs',
'compliance-retention': str(self.compliance_retention_years)
},
# ⚠️ S3 Glacier Instant Retrieval 可进一步降低成本
StorageClass='GLACIER_INSTANT_RETRIEVAL'
)
archived_count = len(batch)
total_size_bytes = len(self._compress_records(batch))
logger.info(f"Archived {archived_count} records to s3://{self.bucket_name}/{s3_key}")
# 删除 Elasticsearch 中的旧数据
# self._delete_es_records("trading-execution-logs", cutoff_date)
except Exception as e:
logger.error(f"Archive failed: {e}")
failed_count = len(batch)
return {
"archived_count": archived_count,
"failed_count": failed_count,
"total_size_bytes": total_size_bytes,
"s3_key": s3_key if archived_count > 0 else None,
"archive_date": archive_date
}
def verify_archive_integrity(self, s3_key: str, expected_record_count: int) -> bool:
"""
验证归档数据的完整性。
监管审计要求能够证明归档数据未被篡改。
"""
try:
response = self.s3.get_object(
Bucket=self.bucket_name,
Key=s3_key,
# 请求数据完整性和校验信息
ChecksumMode='ENABLED'
)
# 检查 S3 元数据中的记录数
metadata_count = int(response['Metadata'].get('record-count', 0))
if metadata_count != expected_record_count:
logger.error(f"Record count mismatch: expected {expected_record_count}, got {metadata_count}")
return False
# 验证压缩数据可以正常解压
import gzip
content = response['Body'].read()
try:
gzip.decompress(content)
except Exception as e:
logger.error(f"Decompression failed: {e}")
return False
logger.info(f"Archive integrity verified: {s3_key}")
return True
except Exception as e:
logger.error(f"Integrity verification failed: {e}")
return False
def generate_compliance_report(self, start_date: str, end_date: str) -> str:
"""
生成合规审计报告。
报告包含指定时间范围内的日志统计、归档状态、数据完整性验证结果。
"""
report = {
"report_id": f"compliance-{start_date}-{end_date}",
"generated_at": datetime.now(timezone.utc).isoformat(),
"period": {
"start": start_date,
"end": end_date
},
"summary": {
"total_execution_logs": 0,
"total_market_data_logs": 0,
"total_strategy_decision_logs": 0,
"archived_size_gb": 0,
"integrity_check_passed": True
},
"archive_details": []
}
# 列出归档桶中的相关文件
paginator = self.s3.get_paginator('list_objects_v2')
for page in paginator.paginate(
Bucket=self.bucket_name,
Prefix=f"execution-logs/year={start_date[:4]}/"
):
if 'Contents' in page:
for obj in page['Contents']:
report['archive_details'].append({
"key": obj['Key'],
"size": obj['Size'],
"last_modified": obj['LastModified'].isoformat(),
"e_tag": obj['ETag'] # S3 ETag 可用于快速校验
})
report['summary']['archived_size_gb'] += obj['Size'] / (1024**3)
return json.dumps(report, indent=2)
def _fetch_es_records(self, index: str, cutoff_date) -> list:
"""从 Elasticsearch 获取待归档记录"""
# ⚠️ 占位实现，需要替换为真实 ES 查询
return []
def _compress_records(self, records: list) -> bytes:
"""压缩日志记录"""
import gzip
content = '\n'.join(json.dumps(r, default=str) for r in records)
return gzip.compress(content.encode('utf-8'))
def _delete_es_records(self, index: str, cutoff_date):
"""删除 Elasticsearch 中的旧记录"""
# ⚠️ 占位实现，需要替换为真实 ES 删除操作
pass
if __name__ == "__main__":
archiver = LogArchiver()
# 归档 7 天前的交易执行日志
result = archiver.archive_execution_logs(days_old=7)
print(f"Archive result: {json.dumps(result, indent=2)}")
# 生成合规报告
report = archiver.generate_compliance_report(
start_date="2024-01-01",
end_date="2024-06-01"
)
print(f"Compliance report: {report}")
4.3 合规数据查询：监管调取响应
当监管机构发出调取令时，需要在规定时间内提供完整的交易重构数据。以下是合规查询模块的示例：
from elasticsearch import Elasticsearch
import boto3
import json
from datetime import datetime, timezone
import hashlib
class ComplianceQuery:
"""
合规数据查询服务。
用于响应监管调取要求，提供完整的交易重构数据。
"""
def __init__(self):
self.es = Elasticsearch([os.environ.get('ES_HOST', 'http://localhost:9200')])
self.s3 = boto3.client('s3')
self.bucket_name = os.environ.get('ARCHIVE_BUCKET', 'trading-compliance-logs')
def reconstruct_trade(self, client_order_id: str, date_range: tuple) -> dict:
"""
还原指定订单的完整交易链路。
Args:
client_order_id: 订单客户端 ID
date_range: 查询日期范围 (start_date, end_date)
Returns:
dict: 完整的交易重构数据
"""
start_date, end_date = date_range
# 1. 查询交易执行日志
execution_logs = self._query_execution_logs(
client_order_id, start_date, end_date
)
# 2. 查询对应的市场数据快照
if execution_logs:
symbol = execution_logs[0]['order']['symbol']
timestamps = [e['timestamp'] for e in execution_logs]
market_data = self._query_market_data(symbol, timestamps)
# 3. 查询策略决策日志
strategy_decisions = self._query_strategy_decisions(
execution_logs[0]['strategy_id'],
timestamps
)
reconstruction = {
"reconstruction_id": f"recon-{client_order_id}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}",
"client_order_id": client_order_id,
"generated_at": datetime.now(timezone.utc).isoformat(),
"execution_timeline": execution_logs,
"market_context": market_data,
"strategy_decisions": strategy_decisions,
"data_sources": ["elasticsearch", "s3-archive"],
"integrity_verified": True
}
return reconstruction
else:
return {"error": "No records found", "client_order_id": client_order_id}
def _query_execution_logs(self, client_order_id: str, start_date: str, end_date: str) -> list:
"""查询 Elasticsearch 中的执行日志"""
query = {
"query": {
"bool": {
"must": [
{"term": {"order.client_order_id": client_order_id}},
{"range": {"timestamp": {"gte": start_date, "lte": end_date}}}
]
}
},
"sort": [{"timestamp": {"order": "asc"}}]
}
result = self.es.search(index="trading-execution-logs-*", body=query, size=1000)
return [hit['_source'] for hit in result['hits']['hits']]
def _query_market_data(self, symbol: str, timestamps: list) -> list:
"""查询订单执行时刻的市场数据"""
# ⚠️ 占位实现
return []
def _query_strategy_decisions(self, strategy_id: str, timestamps: list) -> list:
"""查询策略决策日志"""
# ⚠️ 占位实现
return []
def generate_audit_trail(self, order_ids: list, output_path: str):
"""
生成批量审计追踪报告。
用于响应 SEC/FINRA 等监管机构的调取令。
"""
reconstructions = []
for order_id in order_ids:
recon = self.reconstruct_trade(order_id, (order_id[:8], datetime.now().strftime('%Y-%m-%d')))
reconstructions.append(recon)
report = {
"audit_report_id": f"audit-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}",
"generated_at": datetime.now(timezone.utc).isoformat(),
"total_orders": len(order_ids),
"reconstructions": reconstructions,
"hash": hashlib.sha256(
json.dumps(reconstructions, sort_keys=True, default=str).encode()
).hexdigest()
}
# 写入合规报告文件
with open(output_path, 'w') as f:
json.dump(report, f, indent=2, default=str)
logger.info(f"Audit trail written to {output_path}")
return report
五、数据留存策略的工程实践
5.1 合规留存周期对照
不同监管框架对数据留存周期有明确要求，实际操作中应采用最严格标准：
| 数据类型 | SEC (美国) | MiFID II (欧盟) | 中国 (证监会) | 工程建议 |
|---|---|---|---|---|
| 交易执行记录 | 6 年 | 5 年 | 20 年（期货） | 20 年 |
| 市场数据日志 | 5 年 | 5 年 | 5 年 | 5 年 |
| 系统运行日志 | 3 年 | 3 年 | 3 年 | 3 年 |
| 策略决策日志 | 5 年 | 5 年 | 5 年 | 5 年 |
| 客户账户记录 | 6 年 | 7 年 | 10 年 | 10 年 |
5.2 数据完整性验证机制
合规审计不仅要求数据存在，还要求证明数据未被篡改。推荐以下完整性验证机制：
- 写入时校验：每条日志生成时计算 SHA-256 校验和，随日志一同存储
- 传输加密：Kafka 传输层启用 TLS，存储层使用 S3 SSE-KMS 加密
- 定期哈希树验证：每月对归档数据进行 Merkle 树哈希验证
- 只读存储：冷数据存储使用 S3 Glacier Instant Retrieval 的 Object Lock（WORM）模式
5.3 自动化归档流程
生产环境的日志归档应完全自动化，以下是关键配置要点：
# Kafka Connect S3 Sink Connector 配置示例
{
"name": "trading-logs-s3-sink",
"config": {
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"tasks.max": "3",
"topics": "trading-execution-logs,market-data-logs,strategy-decision-logs",
"s3.bucket": "trading-compliance-logs",
"s3.region": "us-east-1",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"format.class": "io.confluent.connect.s3.format.json.JsonFormat",
"partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
"path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
"timestamp.extractor": "RecordField",
"timestamp.field": "timestamp",
"rotation.schedule.rotation.interval.ms": "3600000",
"schema.compatibility": "NONE"
}
}
此配置实现：
- 按时间分区存储，便于按日期快速定位数据
- 每小时轮转一次，平衡文件大小和查询效率
- JSON 格式存储，支持直接读取和全文检索
六、合规日志体系检查清单
在系统设计评审和合规自查时，可以使用以下检查清单：
6.1 数据完整性
- 每笔订单有唯一的
client_order_id，且不可重复 - 订单状态变更有完整的事件链（不覆盖，只追加）
- 所有日志包含毫秒级或微秒级时间戳
- 所有日志包含来源标识（数据源、系统、策略版本）
- 所有日志包含校验和（SHA-256 或更高）
- 日志格式为结构化 JSON，支持机器解析
6.2 存储与归档
- 热数据层（0-7 天）支持秒级查询
- 温数据层（7-90 天）支持分钟级查询
- 冷数据层满足合规留存周期要求
- S3 存储启用 Object Lock（WORM）模式
- 归档数据每月进行完整性校验
6.3 安全与访问控制
- Kafka 启用 SASL/SSL 认证
- Elasticsearch 启用 RBAC 访问控制
- S3 存储启用 SSE-KMS 加密
- 合规查询操作有独立的审计日志
- 日志访问权限遵循最小权限原则
6.4 灾难恢复
- Kafka 消息持久化到至少 3 个副本
- S3 存储启用跨区域复制（CRR）
- 有完整的备份恢复流程文档
- 定期进行恢复演练
结语
合规审计不是交易的“额外负担”，而是量化系统可信度的基石。一次完整的合规审计能够回答三个问题：这笔交易是否按照策略意图执行、当时的决策是否有量化依据、数据在传输和存储过程中是否保持完整。
当这三个问题都能用日志数据清晰回答时，不仅能满足监管要求，更重要的是——你对自己策略的理解也更加深刻。
下一步行动
如果你正在搭建量化交易系统：
- 将本文的日志结构模板集成到你的交易引擎中
- 部署 Kafka + Elasticsearch + S3 的日志基础设施
- 验证日志完整性校验机制是否生效
如果你需要完整的行情数据支持：
- 访问 tickdb.ai 获取 TickDB Market Data API，支持美股、港股、数字货币的实时行情与历史数据
- 数据格式与本文日志体系天然对齐，可直接用于策略回放与审计验证
如果你希望了解更多合规技术细节：
- SEC Rule 15c3-5 原文：sec.gov/rules/final/2010/34-63545.pdf
- MiFID II transaction reporting：eba.europa.eu/regulation-and-policy/market-activities
风险提示：本文提供的是技术架构指导，不构成任何投资建议。合规要求因司法管辖区和业务类型而异，请咨询专业合规顾问确认具体适用规则。