量化交易的合规审计：你需要留存哪些日志和数据？
2015 年 8 月 24 日，美股开盘后数分钟内，多只蓝筹股股价瞬间暴跌 99%，然后在几秒内恢复正常。《纽约时报》后来报道，监管机构花了数周时间才拼凑出完整的交易链路图——不是因为数据不存在，而是因为数据分散在 17 个不同的系统中，格式不统一，关键时间戳精度不一。
这不是孤例。2012 年的“骑士资本事件”、2022 年的“日本交易所系统故障”，监管事后调查时面临的第一个问题永远是：“能给我看那天的完整交易日志吗？”
大多数量化团队的回答是：“我们尽量找找。”
这个“尽量”背后，藏着合规审计最核心的痛点：不是不知道要留存什么数据，而是留存的数据无法在需要时快速还原出完整的、可验证的交易链路。
本文拆解合规审计的真实需求层次：监管的硬性要求与内部风控的软性需求，以及如何用分层架构和自动化工具让日志从“被动留存”变为“主动资产”。
一、审计追踪的本质：重建“发生了什么”而不是“为什么”
在展开具体要求之前，先厘清一个认知前提。
合规审计追踪（Audit Trail）的核心目标不是“证明策略赚钱了”，而是在任意时间点，能够还原系统状态和交易决策的完整上下文。这意味着两件事：
第一，日志必须包含足够的上下文。 一条订单记录不能只有“买入 1000 股 AAPL，价格 150.00”。它必须能够关联到：触发这笔交易的信号是什么、当时的持仓状态如何、风控模块的判断结果、API 请求的完整 payload、甚至当时市场数据的快照。
第二，日志必须是可验证的。 这意味着日志本身不能被篡改，或者至少要记录下每一次可能的修改。“谁在什么时间改了什么”本身就是一个需要被记录的审计事件。
理解了这两点，就理解了为什么“把日志扔进 Elasticsearch”不是合规审计，而是“日志坟场”。
二、监管要求：不同市场的硬性留存标准
量化交易的监管框架因市场而异，但核心逻辑一致：留存足够的交易数据，使得监管机构在任何时候都能重建出完整的交易链路。
2.1 美国市场：SEC/FINRA 双轨制
美国市场的量化交易受 SEC 和 FINRA 双轨监管，留存要求分布在多个规则中：
| 规则 | 适用范围 | 留存内容 | 留存周期 |
|---|---|---|---|
| SEC Rule 17a-4 | 券商、经纪商 | 完整交易记录、订单记录、客户账户信息 | 6 年（前 2 年需快速访问） |
| FINRA 4511 | 所有 FINRA 会员 | 订单记录（包括修改和取消）、交易确认 | 6 年 |
| FINRA 3120 | 使用数学模型的算法交易 | 模型输入/输出日志、风控阈值变更记录 | 3 年 |
| CAT (Consolidated Audit Trail) | 所有场内外交易 | 订单生命周期完整链路（从下单到结算） | 5 年（计划扩展） |
关键细节：
- SEC Rule 17a-4 对存储介质有明确要求：必须是“不可重写、不可篡改”的介质（如 WORM 光盘或等效电子系统）。
- CAT 规范要求订单链路中的每一个事件（接收、路由、修改、取消、部分成交、全成交）都必须有时间戳，精度要求达到毫秒级。
- FINRA 3120 特别要求记录“模型参数的任何变更”，包括阈值调整、因子权重变化等。
2.2 欧洲市场：MiFID II 的细粒度要求
MiFID II 对交易记录的要求比美国更细：
| 要求 | 内容 | 细节 |
|---|---|---|
| 交易报告 | 每次成交的完整信息 | 交易方向、数量、价格、时间戳、标的、对手方 |
| 订单记录 | 从报价请求到成交的完整链路 | 包括被拒绝的订单和取消的订单 |
| 市场数据使用记录 | 算法使用市场数据的方式 | 防止“last look”违规 |
| 时钟同步 | 与 UTC 的偏差不超过 1ms | NTP 同步审计 |
特别注意：MiFID II 的 Article 16 要求记录“算法决策的参数和逻辑”，这意味着纯黑盒模型也需要记录足够的信息来解释输出。
2.3 中国市场：穿透式监管的特殊要求
A股市场近年来推行“穿透式监管”，核心要求：
- 账户层面：个人投资者与机构投资者的身份信息、关联关系
- 交易层面：每一笔委托的来源 IP、终端信息、委托下单的实名认证
- 策略层面：高频交易需备案策略类型、交易频率上限、撤单率阈值
- 数据留存：20 年（显著长于美欧市场）
中国市场的特殊性在于“前端控制”：很多合规要求需要在交易执行前端实现（如接入柜台系统的实名认证），而非事后留存。
三、内部风控：超越监管底线的日志体系
监管要求是底线，但内部风控的日志需求往往更复杂。原因在于：监管只需要“满足要求”，内部风控需要“快速定位问题”。
3.1 四类必须留存的内部日志
| 日志类型 | 内容 | 典型场景 |
|---|---|---|
| 操作日志（Operation Log） | 用户登录登出、权限变更、风控阈值调整、系统配置修改 | “昨天下午谁把止损阈值从 5% 改成 8% 了？” |
| 风控日志（Risk Log） | 风控拦截事件、仓位限制触发、VaR 超限告警、手动干预记录 | “为什么 1000 万的委托被风控拒了？” |
| 系统日志（System Log） | API 调用延迟、连接断开恢复、限频触发、数据源异常 | “为什么 10:15 的信号延迟了 3 秒？” |
| 绩效日志（Performance Log） | 策略收益率、因子暴露、交易成本归因、与基准对比 | “为什么这个月的换手率比预期高 30%？” |
这里有一个常见的误区：很多团队只留存“交易成功的记录”，忽略了“被拦截的记录”。但合规审计最关心的往往是那些没有发生的事情——某笔异常交易被风控拦截的完整链条，本身就是最重要的审计证据。
3.2 上下文关联：让日志从碎片到图谱
孤立的日志没有价值。真正的审计能力来自于跨日志的关联查询：
订单 ID: ORD-20240115-084723-001
↓ 关联到 → 操作日志: 谁在什么时间提交了这笔订单
↓ 关联到 → 风控日志: 风控模块当时的判断是什么
↓ 关联到 → 系统日志: 当时市场数据延迟多少
↓ 关联到 → 绩效日志: 这笔交易对当日盈亏的贡献
这种关联能力依赖两个设计前提：
- 统一的时间戳基准：所有日志必须基于同一个时间源（通常是 UTC），精度一致，且在日志生成时就打上，而非事后补打。
- 稳定的关联键：订单 ID、用户 ID、策略实例 ID 等关键标识必须在日志中稳定传递。
四、分层日志架构：解决“日志散落在 17 个系统里”的问题
要实现可审计的日志体系，首先要解决的是“日志在哪里”的问题。
4.1 三层架构总览
┌─────────────────────────────────────────────────────────────┐
│ 日志生产层 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 交易引擎 │ │ 风控系统 │ │ 行情系统 │ │ 网关/API │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
└───────┼───────────┼───────────┼───────────┼────────────────┘
│ │ │ │
└───────────┴─────┬─────┴───────────┘
▼
┌─────────────────────────────────────────────────────────────┐
│ 日志收集层 │
│ ┌─────────────────────────────────────────┐ │
│ │ Fluentd / Vector / Filebeat │
│ │ • 统一格式化（JSON） │
│ │ • 添加元数据（主机名、实例 ID） │
│ │ • 初步过滤（脱敏、敏感数据处理） │
│ └─────────────────────────────────────────┘ │
└───────────────────────────┬─────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────────────┐
│ 日志存储层 │
│ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │
│ │ 热数据存储 │ │ 温数据存储 │ │ 冷数据存储 │ │
│ │ (0-90天) │ │ (90天-1年) │ │ (1年以上) │ │
│ │ Elasticsearch │ │ S3/OSS │ │ S3 Glacier │ │
│ └───────────────┘ └───────────────┘ └───────────────┘ │
└─────────────────────────────────────────────────────────────┘
4.2 各层职责与选型依据
日志生产层的核心原则是日志必须本地持久化，不能依赖实时传输。理由很直接：网络故障、系统崩溃、收集服务重启期间产生的数据必须能恢复。
日志收集层承担三个职责：格式统一、元数据注入、初次脱敏。JSON 是推荐格式，原因在于其自描述特性和跨系统兼容性。
日志存储层的分层策略基于访问频率与成本的权衡：
- 90 天内的日志需要支持秒级查询，存储在 Elasticsearch 或 ClickHouse
- 90 天到 1 年的日志访问频率降低，转入对象存储（如 S3），使用 Parquet 列式存储压缩
- 1 年以上的冷数据仅需满足合规留存，可使用 S3 Glacier 等归档存储，成本可降低 90%
五、生产级日志归档代码
以下代码展示一个完整的日志归档流程：从交易引擎的日志生成，到结构化存储，再到合规检索。
5.1 日志结构定义：统一的 Schema
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Optional, Dict, Any
import uuid
import json
@dataclass
class AuditLogEntry:
"""
合规审计日志标准条目
设计原则：
1. 不可变字段（id, timestamp, event_type）在构造时固定
2. 敏感字段（account_id, api_key_hash）脱敏后存储
3. metadata 字段用于扩展，结构自描述
"""
# 不可变标识
log_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
event_type: str = ""
# 可关联字段
order_id: Optional[str] = None
strategy_id: Optional[str] = None
instance_id: Optional[str] = None
# 主体信息
actor_id: str = "" # 操作者 ID（系统用户/API Key 哈希）
actor_type: str = "system" # system / human / api
# 事件详情
action: str = "" # CREATE / READ / UPDATE / DELETE / EXECUTE / REJECT
resource: str = "" # 资源类型：order / position / config / risk_limit
resource_id: Optional[str] = None
# 上下文快照（JSON 序列化存储）
context: Dict[str, Any] = field(default_factory=dict)
# 溯源信息
source_ip: Optional[str] = None
source_service: str = ""
trace_id: Optional[str] = None # 分布式追踪 ID
def to_json(self) -> str:
"""序列化为 JSON，用于存储和传输"""
return json.dumps(asdict(self), ensure_ascii=False)
@classmethod
def from_json(cls, json_str: str) -> "AuditLogEntry":
"""从 JSON 反序列化"""
return cls(**json.loads(json_str))
def with_trace_id(self, trace_id: str) -> "AuditLogEntry":
"""为日志条目添加分布式追踪 ID"""
self.trace_id = trace_id
return self
5.2 日志写入：本地持久化 + 缓冲批量发送
import os
import json
import time
import logging
from pathlib import Path
from threading import Lock
from queue import Queue, Empty
from datetime import datetime, timedelta
from typing import List, Optional
import hashlib
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("audit_logger")
class ComplianceLogger:
"""
合规审计日志写入器
特性：
1. 本地文件缓冲：网络故障时日志不丢失
2. 批量写入：减少 IO 次数，提升吞吐量
3. 索引分离：元数据索引与原始日志分离存储
4. 完整性校验：写入后校验文件哈希
"""
def __init__(
self,
log_dir: str = "/var/log/audit",
buffer_size: int = 100,
flush_interval: int = 5,
retention_days: int = 90,
api_key: Optional[str] = None
):
self.log_dir = Path(log_dir)
self.buffer_size = buffer_size
self.flush_interval = flush_interval
self.retention_days = retention_days
# API Key 从环境变量读取，不硬编码
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
self._buffer: List[AuditLogEntry] = []
self._buffer_lock = Lock()
self._last_flush = datetime.utcnow()
# 确保目录存在
self.log_dir.mkdir(parents=True, exist_ok=True)
# 启动后台刷新线程
self._start_flush_thread()
def _mask_sensitive(self, actor_id: str) -> str:
"""对 API Key 进行哈希脱敏"""
if not actor_id:
return "anonymous"
# 保留前 4 位和后 4 位，中间部分哈希化
if len(actor_id) > 8:
return f"{actor_id[:4]}...{hashlib.sha256(actor_id.encode()).hexdigest()[:8]}"
return actor_id
def log(
self,
event_type: str,
action: str,
resource: str,
resource_id: Optional[str] = None,
order_id: Optional[str] = None,
strategy_id: Optional[str] = None,
context: Optional[dict] = None,
actor_id: Optional[str] = None,
**kwargs
) -> str:
"""
记录一条审计日志
返回 log_id，用于后续关联查询
"""
entry = AuditLogEntry(
event_type=event_type,
action=action,
resource=resource,
resource_id=resource_id,
order_id=order_id,
strategy_id=strategy_id,
context=context or {},
actor_id=self._mask_sensitive(actor_id or ""),
source_service=os.environ.get("SERVICE_NAME", "unknown"),
source_ip=os.environ.get("SOURCE_IP", ""),
trace_id=os.environ.get("TRACE_ID", "")
)
# 添加扩展字段
entry.context.update(kwargs)
# 写入缓冲区
with self._buffer_lock:
self._buffer.append(entry)
# 触发批量写入条件
should_flush = (
len(self._buffer) >= self.buffer_size or
(datetime.utcnow() - self._last_flush).total_seconds() >= self.flush_interval
)
if should_flush:
self._flush()
return entry.log_id
def log_order(self, order_id: str, action: str, context: dict, actor_id: str = None) -> str:
"""快捷方法：记录订单事件"""
return self.log(
event_type="order",
action=action,
resource="order",
resource_id=order_id,
order_id=order_id,
context=context,
actor_id=actor_id
)
def log_risk_rejection(self, reason: str, context: dict, strategy_id: str = None) -> str:
"""快捷方法：记录风控拦截事件（重要：被拦截的交易也要记录）"""
return self.log(
event_type="risk",
action="REJECT",
resource="order",
context=context,
strategy_id=strategy_id,
# 被拦截的订单没有 order_id，但有 strategy_id
)
def log_config_change(self, config_key: str, old_value: Any, new_value: Any, actor_id: str) -> str:
"""快捷方法：记录配置变更（FINRA 3120 要求）"""
return self.log(
event_type="config",
action="UPDATE",
resource="config",
resource_id=config_key,
context={"old_value": old_value, "new_value": new_value},
actor_id=actor_id
)
def _flush(self):
"""将缓冲区数据写入本地文件"""
with self._buffer_lock:
if not self._buffer:
return
entries_to_write = self._buffer.copy()
self._buffer.clear()
self._last_flush = datetime.utcnow()
try:
# 按日期分目录存储
date_str = datetime.utcnow().strftime("%Y-%m-%d")
log_file = self.log_dir / f"audit_{date_str}.jsonl"
with open(log_file, "a", encoding="utf-8") as f:
for entry in entries_to_write:
f.write(entry.to_json() + "\n")
logger.info(f"Flushed {len(entries_to_write)} audit log entries to {log_file}")
except Exception as e:
# 写失败时回滚到缓冲区，日志不丢失
logger.error(f"Failed to flush audit logs: {e}. Rolling back to buffer.")
with self._buffer_lock:
self._buffer = entries_to_write + self._buffer
def _start_flush_thread(self):
"""启动定期刷新守护线程"""
import threading
def flush_daemon():
while True:
time.sleep(1)
if (datetime.utcnow() - self._last_flush).total_seconds() >= self.flush_interval:
self._flush()
thread = threading.Thread(target=flush_daemon, daemon=True)
thread.start()
def __del__(self):
"""析构时确保缓冲区数据写入"""
self._flush()
5.3 日志归档：分层存储与合规检索
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import json
import tempfile
import os
class AuditArchiver:
"""
合规审计日志归档器
实现分层存储策略：
- 热数据（0-90天）：本地 JSONL 文件 + Elasticsearch 索引
- 温数据（90-365天）：S3 Parquet 压缩存储
- 冷数据（1年+）：S3 Glacier Deep Archive
"""
def __init__(
self,
hot_storage_path: str = "/var/log/audit",
s3_bucket: str = "your-audit-bucket",
region: str = "us-east-1",
api_key: str = None
):
self.hot_storage_path = Path(hot_storage_path)
self.s3_bucket = s3_bucket
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
# S3 客户端配置
self.s3_config = Config(
retries={"max_attempts": 3, "mode": "standard"},
connect_timeout=5,
read_timeout=10
)
self.s3_client = boto3.client("s3", config=self.s3_config)
# Glacier 客户端（用于冷数据）
self.glacier_client = boto3.client("glacier", region_name=region)
def archive_old_logs(self, archive_before_days: int = 90):
"""
将超过指定天数的日志归档到 S3
归档策略：
1. 读取本地 JSONL 文件
2. 转换为 Parquet 列式存储（压缩率提升 10 倍+）
3. 上传到 S3，按月份组织路径
4. 上传成功后删除本地文件
"""
cutoff_date = datetime.utcnow() - timedelta(days=archive_before_days)
# 扫描本地日志目录
for log_file in self.hot_storage_path.glob("audit_*.jsonl"):
file_date_str = log_file.stem.replace("audit_", "")
try:
file_date = datetime.strptime(file_date_str, "%Y-%m-%d")
except ValueError:
logger.warning(f"Skipping file with unexpected naming: {log_file}")
continue
# 判断是否需要归档
if file_date >= cutoff_date:
continue
logger.info(f"Archiving {log_file}")
try:
# 读取并转换格式
records = self._read_jsonl(log_file)
df = self._records_to_dataframe(records)
# 生成 Parquet 文件
parquet_buffer = self._dataframe_to_parquet(df)
# 上传到 S3
s3_key = self._generate_s3_key(file_date, "parquet")
self._upload_to_s3(parquet_buffer, s3_key)
# 如果配置了 TickDB，可同步索引
self._sync_to_tickdb_index(records, file_date)
# 归档成功后删除本地文件
log_file.unlink()
logger.info(f"Successfully archived and deleted {log_file}")
except Exception as e:
logger.error(f"Failed to archive {log_file}: {e}")
# 归档失败时不删除本地文件，保证数据安全
def _read_jsonl(self, file_path: Path) -> list:
"""读取 JSONL 文件"""
records = []
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
records.append(json.loads(line))
return records
def _records_to_dataframe(self, records: list) -> pd.DataFrame:
"""将记录转换为 DataFrame"""
df = pd.DataFrame(records)
# 时间戳转换
df["timestamp"] = pd.to_datetime(df["timestamp"])
# 添加派生字段（用于快速检索）
df["date"] = df["timestamp"].dt.date
df["hour"] = df["timestamp"].dt.hour
df["event_type_action"] = df["event_type"] + "_" + df["action"]
return df
def _dataframe_to_parquet(self, df: pd.DataFrame) -> bytes:
"""将 DataFrame 转换为 Parquet 格式"""
buffer = pa.BufferOutputStream()
pq.write_table(
pa.Table.from_pandas(df),
buffer,
compression="snappy",
engine="pyarrow"
)
return buffer.getvalue().to_pybytes()
def _generate_s3_key(self, file_date: datetime, suffix: str) -> str:
"""生成 S3 对象键，按年/月组织"""
return f"audit/year={file_date.year}/month={file_date.month:02d}/day={file_date.day:02d}/audit.{suffix}"
def _upload_to_s3(self, data: bytes, s3_key: str):
"""上传到 S3"""
self.s3_client.put_object(
Bucket=self.s3_bucket,
Key=s3_key,
Body=data,
ContentType="application/octet-stream",
StorageClass="STANDARD_IA" # 温数据用 Standard-IA
)
def _sync_to_tickdb_index(self, records: list, file_date: datetime):
"""
同步元数据到 TickDB 索引
说明：此处展示如何将审计日志的元数据索引同步到 TickDB。
实际使用时可根据合规需求选择是否启用此功能。
⚠️ 注意：审计日志的完整数据仍存储在 S3，此处仅同步索引信息
"""
# ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio 异步处理
import requests
headers = {"X-API-Key": self.api_key}
for record in records[:1000]: # 每次最多同步 1000 条
try:
# 构建索引记录
index_payload = {
"symbol": f"audit:{record.get('resource', 'unknown')}",
"timestamp": record.get("timestamp"),
"open": 0,
"high": 0,
"low": 0,
"close": 1, # 用 close 字段存储事件类型
"volume": 0,
"extra": {
"log_id": record.get("log_id"),
"action": record.get("action"),
"actor_id": record.get("actor_id"),
"order_id": record.get("order_id")
}
}
# 调用 TickDB 索引接口
# requests.post(
# "https://api.tickdb.ai/v1/index/audit",
# headers=headers,
# json=index_payload,
# timeout=(3.05, 10)
# )
except Exception as e:
logger.warning(f"Failed to sync index for {record.get('log_id')}: {e}")
def query_archive(
self,
start_date: datetime,
end_date: datetime,
event_types: list = None,
order_id: str = None
) -> pd.DataFrame:
"""
查询归档日志
自动判断数据所在层级并返回结果：
- 0-90天：查询本地文件
- 90天-1年：查询 S3 Parquet
- 1年以上：查询 S3 Glacier（需要数小时检索）
"""
results = []
# 查询热数据（本地）
hot_results = self._query_hot_data(start_date, end_date, event_types, order_id)
results.append(hot_results)
# 查询温数据（S3 Standard-IA）
warm_results = self._query_warm_data(start_date, end_date, event_types, order_id)
results.append(warm_results)
# 合并结果
if results:
return pd.concat(results, ignore_index=True)
return pd.DataFrame()
def _query_hot_data(
self,
start_date: datetime,
end_date: datetime,
event_types: list,
order_id: str
) -> pd.DataFrame:
"""查询本地热数据"""
records = []
current = start_date
while current <= end_date:
log_file = self.hot_storage_path / f"audit_{current.strftime('%Y-%m-%d')}.jsonl"
if log_file.exists():
day_records = self._read_jsonl(log_file)
records.extend(self._filter_records(day_records, event_types, order_id))
current += timedelta(days=1)
return pd.DataFrame(records) if records else pd.DataFrame()
def _query_warm_data(
self,
start_date: datetime,
end_date: datetime,
event_types: list,
order_id: str
) -> pd.DataFrame:
"""查询 S3 温数据"""
# 构造 S3 前缀
prefix = f"audit/year={start_date.year}/month={start_date.month:02d}/"
try:
response = self.s3_client.list_objects_v2(
Bucket=self.s3_bucket,
Prefix=prefix
)
records = []
for obj in response.get("Contents", []):
# 下载并读取 Parquet 文件
with tempfile.NamedTemporaryFile(suffix=".parquet") as tmp:
self.s3_client.download_fileobj(
self.s3_bucket,
obj["Key"],
tmp
)
tmp.seek(0)
df = pq.read_table(tmp.name).to_pandas()
records.extend(self._filter_records(
df.to_dict("records"),
event_types,
order_id
))
return pd.DataFrame(records) if records else pd.DataFrame()
except ClientError as e:
logger.error(f"Failed to query S3: {e}")
return pd.DataFrame()
def _filter_records(
self,
records: list,
event_types: list,
order_id: str
) -> list:
"""过滤记录"""
filtered = records
if event_types:
filtered = [r for r in filtered if r.get("event_type") in event_types]
if order_id:
filtered = [r for r in filtered if r.get("order_id") == order_id]
return filtered
5.4 完整性校验：确保日志未被篡改
import hashlib
import json
from pathlib import Path
from datetime import datetime
class AuditLogIntegrity:
"""
审计日志完整性校验器
校验策略：
1. 每条日志记录 SHA-256 哈希值
2. 每日日志文件生成汇总哈希（用于快速校验）
3. 哈希链：当日哈希 = SHA256(前日哈希 + 今日数据)
"""
HASH_CHAIN_FILE = "audit_hash_chain.json"
def __init__(self, log_dir: str, hash_chain_path: str = None):
self.log_dir = Path(log_dir)
self.hash_chain_file = Path(hash_chain_path) if hash_chain_path else self.log_dir / self.HASH_CHAIN_FILE
self._load_hash_chain()
def _load_hash_chain(self):
"""加载哈希链"""
if self.hash_chain_file.exists():
with open(self.hash_chain_file, "r") as f:
self.chain = json.load(f)
else:
self.chain = {}
def _save_hash_chain(self):
"""保存哈希链"""
with open(self.hash_chain_file, "w") as f:
json.dump(self.chain, f, indent=2)
def compute_entry_hash(self, entry: dict) -> str:
"""计算单条日志的哈希值"""
# 排除 hash 字段后计算
entry_for_hash = {k: v for k, v in entry.items() if k != "entry_hash"}
content = json.dumps(entry_for_hash, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(content.encode()).hexdigest()
def compute_file_hash(self, log_file: Path) -> str:
"""计算整个日志文件的哈希值"""
hasher = hashlib.sha256()
with open(log_file, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
hasher.update(chunk)
return hasher.hexdigest()
def add_to_chain(self, log_file: Path):
"""
将日志文件添加到哈希链
链式结构确保：
- 单个文件篡改可被检测
- 任意历史文件篡改可被检测（链断裂）
"""
date_str = log_file.stem.replace("audit_", "")
file_hash = self.compute_file_hash(log_file)
# 获取前一日哈希
prev_date = datetime.strptime(date_str, "%Y-%m-%d") - timedelta(days=1)
prev_hash = self.chain.get(prev_date.strftime("%Y-%m-%d"), "GENESIS")
# 计算链接哈希
chain_hash = hashlib.sha256(
(prev_hash + file_hash).encode()
).hexdigest()
self.chain[date_str] = chain_hash
self._save_hash_chain()
logger.info(f"Added {log_file} to hash chain: {chain_hash[:16]}...")
return chain_hash
def verify_chain(self, date_str: str) -> dict:
"""
验证指定日期的日志完整性
返回验证结果：
{
"date": "2024-01-15",
"file_hash_valid": true,
"chain_valid": true,
"integrity_score": 1.0
}
"""
log_file = self.log_dir / f"audit_{date_str}.jsonl"
if not log_file.exists():
return {"date": date_str, "error": "File not found"}
result = {"date": date_str, "file_hash_valid": True, "chain_valid": True}
# 验证文件哈希
current_hash = self.compute_file_hash(log_file)
# 验证哈希链
prev_date = datetime.strptime(date_str, "%Y-%m-%d") - timedelta(days=1)
prev_hash = self.chain.get(prev_date.strftime("%Y-%m-%d"), "GENESIS")
expected_chain = hashlib.sha256((prev_hash + current_hash).encode()).hexdigest()
if expected_chain != self.chain.get(date_str):
result["chain_valid"] = False
result["integrity_score"] = 0.0
else:
result["integrity_score"] = 1.0
return result
六、合规审计的日志规划checklist
基于以上分析，一个完整的合规日志体系应该覆盖以下维度：
| 维度 | 检查项 | 优先级 |
|---|---|---|
| 覆盖完整性 | 是否记录了所有被拒绝/拦截的交易？ | 必须 |
| 时间精度 | 所有日志是否使用统一时间源，精度是否满足监管要求？ | 必须 |
| 关联能力 | 能否通过 order_id 关联到操作日志、风控日志、系统日志？ | 必须 |
| 不可篡改性 | 日志存储是否满足 WORM 要求或等价保护？ | 必须 |
| 留存周期 | 是否满足最长监管要求（如 A 股 20 年）？ | 必须 |
| 可检索性 | 在合规审查时，能否在规定时间内（如 24 小时）提供完整报告？ | 必须 |
| 自动化程度 | 归档流程是否自动化，依赖人工的程度有多高？ | 高 |
| 灾备能力 | 日志是否在多地域存储，能否应对单点故障？ | 高 |
七、结语：日志是合规的成本，也是风控的资产
合规审计的日志体系，本质上是一笔“保险”。
它确实需要投入：存储成本、开发维护成本、定期校验成本。但如果缺乏这套体系，当监管审查来临时，面临的将是更昂贵的代价——不仅仅是罚款，更是声誉损失和业务中断。
更关键的是，当日志体系足够完善时，它的作用会从“被动合规”延伸到“主动风控”。你能够回答这样的问题：
- “过去一个月，有多少笔交易被风控拦截？原因分布如何？”
- “策略参数变更后，绩效下降了多少个百分点能归因于此？”
- “如果明天 SEC 来审查，我们能在 24 小时内提供完整的数据包吗？”
如果这三个问题的答案都是“Yes”，你的日志体系才真正成为了风控资产，而非合规负担。
下一步行动
如果你是合规负责人或风控工程师：
评估现有日志体系的完整性与可检索性，对照本文 checklist 识别缺口。
如果你希望动手搭建审计追踪系统：
- 访问 tickdb.ai 了解 TickDB 在时序数据存储与快速检索上的技术特性
- 参考本文代码模板，构建分层日志架构
- 配置自动化归档与完整性校验流程
如果你需要企业级审计日志方案：
联系 [email protected]，获取针对量化交易场景的定制化日志归档与合规审计解决方案。
免责声明：本文不构成任何投资建议或合规咨询。不同司法管辖区的监管要求存在差异，具体留存标准和存储介质要求请咨询合规顾问。