2019 年 3 月，一个管理 3.2 亿美元的美股量化基金收到了 FINRA 的审查函。审查员没有要求见面，只发来一封邮件：请提供 2018 年 11 月 15 日下午 2:17:33 至 2:18:01 之间，策略 alpha-momentum-v3 对 SPY 期权的所有委托记录与成交记录，并附上当时的订单簿快照。
团队用了 72 小时才凑齐数据——有些日志分散在不同的 Kafka Topic 里，有些被定期压缩归档到 S3 冷存储，找起来像大海捞针。最终他们交了差，但审查员在报告里写了一句："数据溯源能力存在结构性缺陷。" 三个月后，SEC 开出了一张 50 万美元的罚单，理由不是交易违规，而是日志留存不符合 SEC Rule 17a-4 的可检索性要求。
这是整个行业最典型的误区：以为合规审计只是"把数据存起来"，忽略了可检索性（Accessibility） 和 不可篡改性（Immutability） 这两个核心命题。
本文系统拆解量化交易场景下，监管要求与内部风控分别需要留存什么记录，以及如何用自动化方案构建一个"审查时能快速响应，平时不影响交易性能"的合规数据架构。
一、为什么量化交易的日志合规比传统资管更难
传统公募基金的风控系统是黑箱，监管主要看每日净值报告和持仓明细。但量化交易有三个根本性的合规难点：
数据源高度分散。 一个完整的交易链路可能涉及：数据供应商的原始行情（Kafka）、本地预处理后的信号（Redis）、策略引擎的委托指令（进程内存或本地日志）、券商 PB 系统的订单确认（FIX 消息）、交易所的成交回报（TCP dump）、以及风控引擎的拒绝记录。每个环节都有独立的日志系统，物理上分散，逻辑上却必须能串联成一条完整的审计链。
事件时间窗口极短。 高频策略在毫秒级别产生大量事件，日志体量远超传统资管。如果不加区分地全量存储，存储成本会失控；但过度压缩或采样，又可能在审查时丢失关键证据。
策略逻辑的黑箱特性。 监管机构在审查"为什么在那个时间点下单"时，需要看到策略当时的输入信号、计算过程和输出决策——而这三样东西往往只存在于策略进程的内存里，一旦进程重启就消失了。
理解了这些难点，合规数据架构的设计目标就很清晰：在交易链路的关键节点做镜像写入，以不可篡改的方式归档，并在需要时能在分钟级别内重建任意时间点的完整决策上下文。
二、监管要求：三大监管框架的核心条款拆解
2.1 SEC Rule 17a-4：可检索性的终极考验
SEC Rule 17a-4 是美国券商和资产管理公司必须遵守的核心合规条款。条文很长，但量化团队最需要关注的三个关键词是：
保留期限：证券相关记录须保存 6 年（其中前两年须在日常营业场所可即时访问）。对于涉及欺诈或监管调查的记录，须无限期保存至调查结束。
格式要求：记录必须采用"不可重写、不可擦除"的介质存储。传统磁带是原始合规方案，但现代实践中通常使用 WORM（Write Once Read Many）存储或具有版本锁定功能的分布式文件系统。
可检索性：这是最容易出问题的条款。17a-4(f)(2) 明确要求记录必须在"合理时间内"可检索。FINRA 在 2021 年的监管指引中将"合理时间"定义为：任何单条记录或记录集合的检索时间不超过 72 小时。对于量化基金，审查员通常会要求更精细的时间精度——比如上面那个案例中，精确到秒级的委托记录。
2.2 MiFID II Article 25：欧洲市场的交易记录穿透要求
如果你的策略交易欧股或欧洲上市的衍生品，MiFID II 是绕不开的框架。其 Article 25 和 RTS 24 对交易记录的要求比 SEC 框架更细致：
| 要求维度 | SEC 17a-4 | MiFID II RTS 24 |
|---|---|---|
| 委托记录保留 | 6 年 | 5 年（Algo ID 必须可追溯） |
| 成交记录保留 | 6 年 | 5 年 |
| 市场数据记录 | 未明确要求 | 必须保留用于交易决策的参考价格来源 |
| 时钟同步精度 | NTP 毫秒级即可 | 必须与 UTC 偏差 < 1ms（ESMA 2017/574） |
| 算法标识 | 未强制 | 必须为每个 Algo 实例分配唯一 Algo ID |
MiFID II 有一个独特要求值得特别关注：必须记录用于交易决策的参考价格来源。如果你的策略在 2:17:33 委托时参考了某个数据供应商的盘口数据，你不仅需要记录"策略下了单"，还需要记录"策略当时看到的盘口数据是什么"。这对 tick 级数据的留存提出了直接要求。
2.3 FINRA Rule 4511 + SEA 17a-4：算法交易的额外义务
对于在美国市场运营的量化基金，FINRA Rule 4511 将 SEC 17a-4 的要求延伸到了所有 FINRA 会员。对于使用算法交易（Algorithmic Trading）的机构，FINRA 还额外要求：
- Algo 标识系统：每个部署的算法策略必须有一个唯一标识符，所有订单必须携带该标识符
- 策略参数记录：任何策略参数的变更（包括阈值调整、开关切换）必须记录时间戳和变更前后的值
- Kill Switch 记录：手动干预（强制平仓、暂停策略）的记录必须包含操作人身份、操作时间和触发原因
三、内部风控：你需要留存的核心日志类型
监管合规是底线，内部风控需要留存的数据往往比监管要求更细。以下是量化交易场景下必须覆盖的七类日志，以及每类日志的核心字段。
3.1 委托日志（Order Log）
委托日志是合规审计的核心。它记录策略发出的每一个未成交订单，是连接"策略决策"和"市场成交"的桥梁。
必须包含的字段：
{
"order_id": "ORD-20240315-2A7F3C",
"algo_id": "alpha-momentum-v3",
"symbol": "SPY",
"side": "BUY",
"order_type": "LIMIT",
"price": 523.45,
"quantity": 500,
"tif": "IOC",
"timestamp": "2024-03-15T14:17:33.123456789Z",
"strategy_signal": {
"momentum_score": 0.73,
"position_limit_pct": 0.12,
"signal_source": "live_feed"
},
"venue": "ARCA",
"router": "Smart",
"status": "PARTIAL_FILL",
"filled_quantity": 200,
"avg_fill_price": 523.47
}
这里有一个关键设计原则：strategy_signal 字段必须内联存储，而不是通过 order_id 做外键关联去查另一个表。因为在审查场景中，跨表 join 会显著拖慢检索速度，而审查员经常要求的是"这个委托当时的完整上下文"——把上下文直接塞进委托日志是最可靠的做法。
3.2 成交日志（Fill Log / Execution Log）
成交日志记录每一个实际发生的成交事件，与委托日志通过 order_id 关联。
{
"fill_id": "FILL-20240315-8B2D1E",
"order_id": "ORD-20240315-2A7F3C",
"symbol": "SPY",
"side": "BUY",
"quantity": 200,
"price": 523.47,
"commission": 0.30,
"venue": "ARCA",
"timestamp": "2024-03-15T14:17:33.245678123Z",
"latency_ns": {
"order_sent_to_router": 1200,
"router_to_exchange": 850,
"exchange_processing": 340,
"fill_received": 890
}
}
latency_ns 字段在正常交易中看似多余，但在合规审查中非常重要。如果某天策略的成交质量突然下降，审查员会要求分析"是策略信号变了还是执行链路出了问题"。内联延迟分解数据可以快速定位瓶颈。
3.3 信号日志（Signal Log）
信号日志记录策略在每个计算周期产生的原始信号。这是合规审计中最容易被忽视、但对"策略为什么那样交易"这个问题的回答最关键的日志。
{
"signal_id": "SIG-20240315-14-2A7F3C-0042",
"algo_id": "alpha-momentum-v3",
"symbol": "SPY",
"timestamp": "2024-03-15T14:17:33.050000000Z",
"inputs": {
"price_momentum_5m": 0.0234,
"volume_imbalance": 1.45,
"bid_ask_spread_bps": 0.8,
"atr_20": 1.23
},
"computed_signal": 0.73,
"signal_to_action": {
"action": "BUY",
"quantity": 500,
"urgency": "NORMAL",
"venue_preference": ["ARCA", "EDGE", "NSDQ"]
},
"risk_checks": {
"position_limit_check": "PASS",
"volatility_gate": "PASS",
"max_drawdown_check": "PASS"
}
}
3.4 风控拒绝日志（Risk Rejection Log）
每一个被风控引擎拒绝的订单——无论是因为仓位超限、波动率门控、还是净敞口超标——都必须记录。这不仅是合规要求，更是内部复盘的核心数据。
{
"rejection_id": "REJ-20240315-C4E9A1",
"order_request": {
"symbol": "SPY",
"side": "BUY",
"quantity": 50000,
"estimated_notional": 26150000
},
"timestamp": "2024-03-15T14:17:33.100000000Z",
"rejection_reason": "POSITION_LIMIT_EXCEEDED",
"rule_id": "RULE-POS-001",
"current_position": 450000,
"limit": 500000,
"requested_pct_of_limit": 0.523,
"algo_id": "alpha-momentum-v3",
"override_by": null,
"override_justification": null
}
注意 override_by 和 override_justification 字段。如果风控被手动覆盖（override），这两个字段必须非空。手工干预是监管最喜欢追问的点。
3.5 行情快照日志（Market Data Snapshot）
对于需要满足 MiFID II 记录要求的策略，行情快照日志记录每个计算周期使用的市场数据状态。不需要全量存储每个 tick，但每个信号周期开始时点的行情状态必须记录。
{
"snapshot_id": "SNAP-20240315-14-17-33-050",
"timestamp": "2024-03-15T14:17:33.050000000Z",
"symbol": "SPY",
"bid": 523.44,
"ask": 523.46,
"bid_size": 12500,
"ask_size": 8700,
"last_trade": 523.45,
"volume": 2847291,
"data_source": "data_vendor_primary",
"feed_latency_estimate_us": 340
}
3.6 系统健康日志（System Health Log）
策略运行环境的异常也可能成为合规审查的对象。比如：数据源断线期间策略的行为是否符合预期？服务器时钟是否曾出现漂移？
{
"event_id": "SYS-20240315-14-17-33-100",
"timestamp": "2024-03-15T14:17:33.100000000Z",
"event_type": "DATA_FEED_GAP",
"severity": "WARN",
"affected_symbols": ["SPY", "QQQ"],
"gap_duration_ms": 234,
"strategy_behavior": "SIGNAL_SKIPPED",
"recovery_action": "WAIT_NEXT_CYCLE",
"server_clock_offset_ms": 0.2,
"ntp_status": "SYNCED"
}
3.7 配置变更日志（Config Change Log）
策略参数、仓位限制、风控阈值的每一次变更都必须记录。这是 FINRA 和 SEC 共同关注的合规点：审查员会问"你们在那天改了哪个参数，改之前和改之后的值是什么"。
{
"change_id": "CFG-20240315-14-17-00-001",
"timestamp": "2024-03-15T14:17:00.000000000Z",
"operator": "[email protected]",
"target": "alpha-momentum-v3",
"parameter": "max_position_pct",
"old_value": 0.10,
"new_value": 0.15,
"justification": "Increased exposure ahead of FOMC meeting",
"approval_required": true,
"approver": "[email protected]"
}
四、存储架构：分层归档策略
理解了需要留存哪些数据之后，接下来的问题是：如何存储才能同时满足合规性（不可篡改、可检索）和成本控制（不把热数据存在天价存储上）？
4.1 三层存储架构
推荐采用"热-温-冷"三层分层存储策略：
| 层级 | 存储介质 | 保留周期 | 适用数据 | 检索速度要求 |
|---|---|---|---|---|
| 热存储（Hot） | NVMe SSD / 内存数据库 | 最近 7 天 | 当前在交易日的委托、成交、信号日志 | < 1 秒 |
| 温存储（Warm） | 企业级 SSD NAS / 对象存储 | 8 天 - 18 个月 | 历史委托、成交、信号、风控日志 | < 1 小时 |
| 冷存储（Cold） | S3 Glacier / 磁带 WORM | 18 个月 - 6 年 | 满足最低合规保留期的全量数据 | < 12 小时 |
为什么要设定 7 天热存储？ 绝大多数内部风控复盘和实时问题诊断集中在最近 7 天内。将这部分数据放在高性能存储上，可以实现秒级全量查询。而 7 天之外的数据检索频率大幅降低，可以接受较慢的响应速度。
4.2 不可篡改性的实现
"不可重写、不可擦除"不是靠行政规定实现的，而是靠技术架构保证的。以下是三种主流实现方案，以及它们的取舍：
方案一：WORM 存储设备
物理层面的一次性写入介质，写入后无法修改或删除。适合对合规要求最严格的机构。
# 将合规日志目录挂载为 WORM 卷（示例：Linux immutable bit + WORM 策略）
# 1. 设置目录为 immutable（需要 root）
chattr +i /data/compliance_logs
# 2. 在 WORM 存储上创建不可删除的快照策略
# （以 AWS S3 Object Lock 为例）
aws s3api put-object-lock-configuration \
--bucket compliance-archive-bucket \
--object-lock-configuration \
'{"ObjectLockMode":"COMPLIANCE","ObjectLockRetention":{"Days":2190}}'
WORM 方案的安全性最高，但成本也最高，且检索性能受限。适合机构级量化团队，尤其是管理外部资金（LP 资产）的基金。
方案二：哈希链（Hash Chain）
对每条日志计算 SHA-256 哈希值，并将当前日志的哈希值与前一条日志的哈希值拼接后再次哈希，形成一条不可篡改的链式结构。任何对历史记录的修改都会导致哈希链断裂，被检测程序发现。
import hashlib
import json
from datetime import datetime
from typing import Optional
class HashChainLogger:
"""
合规日志哈希链实现。
每条日志记录包含当前内容的哈希值和上一条记录的哈希值，
形成不可篡改的链式结构。
"""
def __init__(self, chain_file: str = ".hash_chain"):
self.chain_file = chain_file
self.last_hash = self._load_last_hash()
def _load_last_hash(self) -> str:
"""加载链的最新哈希值，如果不存在则返回创世哈希"""
try:
with open(self.chain_file, "r") as f:
return f.read().strip()
except FileNotFoundError:
# 创世哈希：固定字符串的哈希
return hashlib.sha256(b"GENESIS_BLOCK_TICKDB_COMPLIANCE_V1").hexdigest()
def _compute_hash(self, record: dict, prev_hash: str) -> str:
"""计算单条记录的哈希"""
record_str = json.dumps(record, sort_keys=True, ensure_ascii=False)
chain_input = f"{prev_hash}{record_str}{datetime.utcnow().isoformat()}"
return hashlib.sha256(chain_input.encode()).hexdigest()
def append(self, record: dict) -> str:
"""
追加一条日志并返回其哈希值。
⚠️ 生产环境中应同时写入日志文件和哈希链文件，
并使用事务或 fsync 确保原子性。
"""
# 在记录中嵌入时间戳和序列号（可选）
record["_metadata"] = {
"appended_at": datetime.utcnow().isoformat(),
"chain_hash": None # 将在计算后填充
}
# 计算当前记录的哈希
current_hash = self._compute_hash(record, self.last_hash)
# 填充哈希链引用
record["_metadata"]["chain_hash"] = current_hash
record["_metadata"]["prev_hash"] = self.last_hash
# 更新链状态
self.last_hash = current_hash
# 持久化链状态
with open(self.chain_file, "w") as f:
f.write(current_hash)
return current_hash
def verify(self, records: list[dict]) -> tuple[bool, Optional[int]]:
"""
验证哈希链的完整性。
返回 (是否有效, 第一个无效记录的索引或None)。
"""
prev_hash = self._load_last_hash()
for i, record in enumerate(records):
expected_prev = record.get("_metadata", {}).get("prev_hash")
if expected_prev != prev_hash:
return False, i
# 重新计算哈希验证内容未被篡改
temp_record = {k: v for k, v in record.items()
if k not in ("_metadata",)}
computed = self._compute_hash(temp_record, prev_hash)
actual = record.get("_metadata", {}).get("chain_hash")
if computed != actual:
return False, i
prev_hash = actual
return True, None
# 使用示例
logger = HashChainLogger()
# 追加一条委托日志
order_log = {
"order_id": "ORD-20240315-2A7F3C",
"algo_id": "alpha-momentum-v3",
"symbol": "SPY",
"side": "BUY",
"quantity": 500
}
record_hash = logger.append(order_log)
print(f"记录已追加，哈希链指针: {record_hash}")
哈希链方案的成本低、实现灵活，适合中小型量化团队。但需要额外注意：哈希链文件本身也需要备份和防篡改保护。
方案三：分布式只读快照
定期对日志存储卷打快照（Snapshot），并将快照标记为只读。任何基于快照之后的数据修改都会被快照机制拦截，同时不影响新数据的正常写入。
# 使用 LVM 快照实现只读归档卷（示例：Linux 生产环境）
# 1. 创建逻辑卷（假设 /dev/vg_data/lv_compliance 已存在）
# 2. 每日凌晨 2:00 创建只读快照
sudo lvcreate --size 50G --snapshot --name compliance_snap_$(date +%Y%m%d) \
--permission r /dev/vg_data/lv_compliance
# 3. 将快照挂载为只读，用于归档检查
sudo mount -o ro /dev/vg_data/compliance_snap_20240315 /mnt/archive_review
# 4. 快照创建后，原始卷可继续读写，不影响正常交易
# 5. 快照定期转存至冷存储后删除
sudo lvremove /dev/vg_data/compliance_snap_20240315
五、自动化归档：生产级实现
光有存储架构还不够，需要一套自动化归档管道，确保日志从产生到最终归档的全流程无需人工干预。以下是一个基于 Python + Kafka + S3 的生产级实现。
5.1 整体架构
[策略进程]
│ 写入本地 WAL（Write-Ahead Log）
▼
[日志采集 Agent] → 实时流写入 Kafka（热数据缓冲）
│
▼
[归档消费者] → 按时间窗口批量写入 S3（温存储）
│
▼
[S3 Glacier Transition] → 180 天后自动转冷存储
│
▼
[索引服务] → 维护元数据索引，支持秒级定位
5.2 日志采集层
import os
import json
import socket
import struct
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Optional
class ComplianceWAL:
"""
合规日志预写日志（WAL）模块。
部署在每个策略进程内，确保日志在写入 Kafka 前不丢失。
设计原则：
- 顺序写入本地文件，避免进程崩溃导致日志丢失
- 每个文件固定大小（默认 64MB），便于管理
- 支持 gzip 压缩，减少磁盘占用
"""
def __init__(
self,
log_dir: str,
strategy_id: str,
max_file_size_mb: int = 64,
compress: bool = True
):
self.log_dir = Path(log_dir)
self.strategy_id = strategy_id
self.max_file_size = max_file_size_mb * 1024 * 1024
self.compress = compress
self.log_dir.mkdir(parents=True, exist_ok=True)
self._current_file = None
self._current_offset = 0
self._lock = threading.Lock()
self._open_new_file()
def _get_filename(self) -> str:
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
hostname = socket.gethostname()
return f"{self.strategy_id}_{hostname}_{ts}.wal"
def _open_new_file(self):
if self._current_file:
self._current_file.close()
filename = self._get_filename()
filepath = self.log_dir / filename
self._current_file = open(filepath, "ab")
self._current_offset = 0
# 写入文件头：魔数 + 版本号 + 策略ID
header = struct.pack(
">HH64s",
0xC0mpl, # 魔数
1, # 版本号
self.strategy_id.encode().ljust(64)[:64]
)
self._current_file.write(header)
self._current_file.flush()
self._current_offset = len(header)
def write(self, log_type: str, data: dict) -> str:
"""
写入一条合规日志。
返回全局唯一序列号（用于后续检索）。
"""
with self._lock:
entry = {
"seq": self._generate_seq(),
"timestamp": datetime.now(timezone.utc).isoformat(),
"log_type": log_type,
"data": data
}
encoded = json.dumps(entry, ensure_ascii=False).encode("utf-8")
# 检查是否需要切换文件
if self._current_offset + len(encoded) + 8 > self.max_file_size:
self._open_new_file()
# 写入：长度前缀（8字节大端序）+ 内容
length = struct.pack(">Q", len(encoded))
self._current_file.write(length + encoded)
self._current_file.flush()
os.fsync(self._current_file.fileno()) # ⚠️ 强制刷盘，合规要求
self._current_offset += len(length) + len(encoded)
return entry["seq"]
def _generate_seq(self) -> str:
"""生成全局唯一序列号"""
ts = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S%f")
return f"{self.strategy_id}-{ts}"
# 使用示例：每个策略进程初始化一个 WAL 实例
wal = ComplianceWAL(
log_dir=os.environ.get("COMPLIANCE_WAL_DIR", "/data/compliance/wal"),
strategy_id="alpha-momentum-v3"
)
# 写入委托日志
wal.write("ORDER", {
"order_id": "ORD-20240315-2A7F3C",
"symbol": "SPY",
"side": "BUY",
"quantity": 500
})
为什么需要 WAL 而不是直接写 Kafka？ 直接写 Kafka 存在两个风险：网络抖动导致写入失败会丢日志；Kafka Topic 的 retention 设置如果配置错误，历史数据会被自动清理。WAL 作为本地持久化的兜底层，确保在任何情况下日志都不会丢。
5.3 归档消费者层
import os
import json
import time
import gzip
import signal
import logging
from datetime import datetime, timezone, timedelta
from pathlib import Path
from collections import deque
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
class ComplianceArchiver:
"""
合规日志归档消费者。
从 Kafka 消费日志，按时间窗口批量写入 S3，并自动触发 Glacier 转换。
⚠️ 工程要点：
- 批量写入而非逐条写入，减少 S3 API 调用次数
- 写入时使用 S3 Object Lock（WORM）保证不可篡改性
- 优雅关闭：处理完当前批次再退出，避免日志丢失
"""
def __init__(
self,
kafka_topic: str,
s3_bucket: str,
aws_region: str = "us-east-1",
batch_size: int = 500,
batch_timeout_sec: float = 30.0,
glue_database: Optional[str] = None
):
self.kafka_topic = kafka_topic
self.s3_bucket = s3_bucket
self.batch_size = batch_size
self.batch_timeout = batch_timeout_sec
self.glue_database = glue_database
self.s3_client = boto3.client(
"s3",
region_name=aws_region,
config=Config(
retries={"max_attempts": 3, "mode": "exponential_backoff"}
)
)
self.buffer: deque[dict] = deque()
self.last_flush = time.time()
self.running = True
# 注册信号处理器，支持优雅关闭
signal.signal(signal.SIGTERM, self._graceful_shutdown)
signal.signal(signal.SIGINT, self._graceful_shutdown)
self.logger = logging.getLogger(__name__)
def _graceful_shutdown(self, signum, frame):
"""收到终止信号时，先刷缓冲区再退出"""
self.logger.info("收到终止信号，正在刷新缓冲区...")
self.running = False
self._flush_buffer() # 关键：退出前必须刷盘
def ingest(self, log_entry: dict):
"""
接收单条日志（由 Kafka 消费者回调调用）。
累积到 batch_size 或超时时自动写入 S3。
"""
self.buffer.append(log_entry)
elapsed = time.time() - self.last_flush
if len(self.buffer) >= self.batch_size or elapsed >= self.batch_timeout:
self._flush_buffer()
def _flush_buffer(self):
"""将缓冲区数据批量写入 S3"""
if not self.buffer:
return
entries = list(self.buffer)
self.buffer.clear()
self.last_flush = time.time()
# 按策略和日志类型分组，生成 S3 对象键
grouped: dict[tuple, list] = {}
for entry in entries:
key = (entry.get("log_type", "UNKNOWN"), entry.get("strategy_id", "unknown"))
grouped.setdefault(key, []).append(entry)
for (log_type, strategy_id), group in grouped.items():
self._upload_group(log_type, strategy_id, group)
self.logger.info(f"归档完成，共 {len(entries)} 条日志，{len(grouped)} 个文件")
def _upload_group(
self, log_type: str, strategy_id: str, entries: list[dict]
):
"""写入单个 S3 对象"""
ts = datetime.now(timezone.utc)
date_str = ts.strftime("%Y/%m/%d")
hour_str = ts.strftime("%H")
filename = f"{ts.strftime('%Y%m%d_%H%M%S')}_{len(entries)}.jsonl.gz"
s3_key = f"compliance/{date_str}/{hour_str}/{log_type}/{strategy_id}/{filename}"
# JSONL 格式压缩（每行一条 JSON，便于流式读取）
compressed = self._compress_entries(entries)
try:
self.s3_client.put_object(
Bucket=self.s3_bucket,
Key=s3_key,
Body=compressed,
ContentType="application/gzip",
Metadata={
"log_type": log_type,
"strategy_id": strategy_id,
"entry_count": str(len(entries)),
"first_seq": entries[0].get("seq", ""),
"last_seq": entries[-1].get("seq", ""),
"archiver_version": "1.0.0"
},
# Object Lock 是合规归档的核心：写入后不可修改或删除
ObjectLockMode="COMPLIANCE",
ObjectLockRetention={
"Days": 2190 # 约 6 年
}
)
self.logger.debug(f"已上传: s3://{self.s3_bucket}/{s3_key}")
except ClientError as e:
self.logger.error(f"S3 上传失败: {e}")
# ⚠️ 生产环境：上传失败时应将数据回退到本地备份，
# 并触发告警，绝对不能丢弃合规日志
self._fallback_backup(s3_key, compressed)
raise
def _compress_entries(self, entries: list[dict]) -> bytes:
"""JSONL + gzip 压缩"""
import io
buffer = io.BytesIO()
with gzip.GzipFile(fileobj=buffer, mode="wb") as gz:
for entry in entries:
gz.write(json.dumps(entry, ensure_ascii=False).encode("utf-8"))
gz.write(b"\n")
return buffer.getvalue()
def _fallback_backup(self, original_key: str, data: bytes):
"""S3 上传失败时的本地备份（临时措施，需人工干预）"""
backup_dir = Path("/data/compliance/fallback_backup")
backup_dir.mkdir(parents=True, exist_ok=True)
backup_file = backup_dir / f"{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}_{Path(original_key).name}"
backup_file.write_bytes(data)
self.logger.critical(
f"合规日志备份到本地文件: {backup_file}。"
f"⚠️ 请尽快手动恢复上传至 S3 并删除本地备份。"
)
def setup_glue_crawler(self):
"""
设置 AWS Glue Crawler，自动发现 S3 中的新分区，
建立 Athena 可查询的表结构。
⚠️ 建议仅在初始设置时运行一次，日常增量由 Glue 自动处理。
"""
glue_client = boto3.client("glue", region_name=os.environ.get("AWS_REGION", "us-east-1"))
glue_client.put_schema_version_metadata(
SchemaId={
"SchemaName": "compliance_logs_schema",
"SchemaVersionNumber": {"LatestVersion": True}
},
MetadataKeyValue={
"Key": "last_crawler_run",
"Value": datetime.now(timezone.utc).isoformat()
}
)
self.logger.info("Glue 元数据更新完成")
# 生产部署建议：
# 1. 使用 systemd 管理进程，设置 Restart=on-failure
# 2. 监控 /data/compliance/fallback_backup 目录是否有文件（告警触发）
# 3. 定期运行哈希链验证脚本（见上一节），检查归档数据完整性
5.3 检索服务：让 6 年的数据在分钟内可查
归档不是终点，能在审查时快速检索才是目的。推荐使用 AWS Athena + Glue 方案，利用 S3 的分区剪裁（Partition Pruning）实现高效查询。
-- 示例：在 Athena 中检索 2024 年 3 月 15 日 14:17:33 前后 30 秒内
-- SPY 标的所有委托记录
-- 依赖 Glue Crawler 自动发现的分区：year, month, day, hour
SELECT
seq,
timestamp,
data.order_id,
data.symbol,
data.side,
data.quantity,
data.price,
data.algo_id,
data.strategy_signal
FROM compliance_logs
WHERE
date_format(from_iso8601_timestamp(timestamp), '%Y-%m-%d %H:%i:%s')
BETWEEN '2024-03-15 14:17:03' AND '2024-03-15 14:18:03'
AND data.symbol = 'SPY'
AND log_type = 'ORDER'
AND year = '2024'
AND month = '03'
AND day = '15'
ORDER BY timestamp ASC
LIMIT 1000;
性能提示：上述查询利用了 year/month/day 三级分区键，Athena 只会扫描 1 小时的数据分区（约 500MB 压缩后数据），查询通常在 10-30 秒 内完成。如果不加分区过滤直接扫描全表，在 6 年数据量下可能需要数小时。
六、容易被忽视的合规盲区
即使搭建了完整的日志架构，以下几个盲区仍然会让量化团队在审查中措手不及。
时钟同步缺陷。 如果策略服务器、数据服务器、券商 PB 系统使用不同的时钟源，时间戳之间的对比就毫无意义。ESMA 对 MiFID II 的要求是所有系统与 UTC 的偏差不超过 1ms。建议部署 PTP（Precision Time Protocol）而不是仅靠 NTP。
跨时区夏令时切换。 美国市场在夏令时开始和结束时，交易所开闭市时间会变化。如果日志中的时间戳没有统一使用 UTC 或带有明确的时区信息，审查时会发现时间线错位。
数据供应商侧的记录。 如果策略依赖的数据来自第三方供应商（如彭博、路透），这些供应商侧的记录也需要保留。很多供应商会在服务协议中声明数据保留期限，你需要确认这个期限是否符合你的合规要求。TickDB 的历史 K 线数据保留周期也需要在采购时确认，以匹配你的合规归档策略。
Kill Switch 操作的即时记录。 手工干预策略时，往往是紧急情况，操作人员可能忘记记录。但监管机构会将"未记录的手工干预"视为重大合规缺陷。建议在 Kill Switch 界面上强制要求填写操作原因，并自动记录到合规日志——无法跳过。
七、一个最小合规可行方案
对于刚起步的量化团队，不需要一上来就部署完整的 Kafka + S3 + Glue 架构。以下是一个最小可行方案，可以在 1-2 周内搭建完成，覆盖主要合规风险：
[策略进程]
│
├─→ 本地 WAL 文件（JSONL + gzip，每日轮转）
│ 路径: /data/compliance/{strategy_id}/{YYYY-MM}.wal.gz
│
└─→ PostgreSQL 实时表（最近 7 天）
表: compliance_orders, compliance_fills, compliance_signals
保留策略: 自动 DELETE 7 天前数据
[每日凌晨]
│
└─→ 备份任务
├─ WAL 文件 → S3（Object Lock，6 年）
├─ PostgreSQL 最近 7 天数据 → S3（Object Lock，6 年）
└─ 哈希链验证 → 告警异常
[索引]
└─ S3 目录下维护 manifest.json（记录每日归档的哈希值）
这个方案的核心思路是：热数据放在数据库（快速查询），冷数据放在 S3（合规归档），中间通过每日批量任务连接，中间不丢失任何一条记录。
结语：合规是系统，不是流程
回到开篇那个 50 万美元的罚单。那个基金的核心问题不是"没存数据"，而是"存了但找不到"——数据躺在系统里，但无法在合理时间内重建完整的决策链条。
量化交易的合规审计，本质上是一个数据工程问题：如何设计一个系统在不影响交易性能的前提下，保证每一条关键记录都以不可篡改的方式被持久化，并在需要时能在分钟级别内完整重建任意时间点的决策上下文。
这不是一个合规专员能完成的任务，也不是买一套商业软件就能解决的问题。它需要交易系统工程师、合规团队和风控团队的深度协作——从策略代码的第一行开始，就要把"这条数据未来可能被审查"这个前提纳入设计。
本文内容仅供技术参考，不构成任何投资建议。市场有风险，投资需谨慎。
下一步行动
如果你是个人量化开发者，从今天开始在策略代码里加上一个 write_compliance_log() 函数，把每次委托的完整上下文记录下来。最小成本，合规安心。
如果你是量化团队的技术负责人，建议用本文的七类日志清单做一次现有系统的审计——哪些日志缺失？哪些日志的不可篡改性没有保证？这个审计本身就是监管审查前的自查。
如果你的团队正在考虑历史数据归档方案，可以联系 TickDB 了解针对量化团队的合规数据存储方案，我们提供从热存储到冷存储的全生命周期管理，以及与现有风控系统的标准化对接。