第15章合规与审计

15、合规与审计：交易记录存储，监管报告生成（MiFID II/Dodd-Frank），反洗钱（AML）检查

做量化做市，技术再牛，合规不过关，一切都白搭。这不是危言耸听，我在一家欧洲的做市商公司待过，亲眼见过一个交易团队因为日志记录不规范，被监管罚了上百万欧元。说白了，合规不是束缚，而是做市商业务的「安全气囊」。

今天这一章，我们就来聊聊做市商系统里最「枯燥」但又最「致命」的三个模块：交易记录怎么存、监管报告怎么出、反洗钱怎么查。嗯，我会尽量讲得接地气一点。

核心观点：合规系统不是事后补丁，而是交易流水线的一部分。数据一旦丢失或格式不对，你连解释的机会都没有。

15.1 交易记录存储：不只是写日志那么简单

很多人觉得，交易记录不就是把订单、成交、撤单写进数据库吗？其实没那么简单。监管要求的是「完整、不可篡改、可追溯」的记录链。

我个人习惯把交易记录分为三层：

原始事件层：交易所发来的每一个消息，包括行情、订单确认、成交回报、拒绝原因。这些是「证据」。
业务对象层：订单、成交、持仓、资金变动。这些是「事实」。
审计快照层：每隔一段时间（比如1分钟）对系统状态做一次全量快照。这些是「时间胶囊」。

为什么要做快照？我在项目中遇到过一个问题：某个订单在数据库里被更新了多次，但中间某个状态丢了。监管问「这个订单为什么在10:00:03变成了已撤销？」你拿不出证据。有了快照，你就能说「看，10:00:00的时候它还是活的，10:01:00它已经撤销了，中间发生了什么？这里有事件日志。」

我的建议：存储引擎选时序数据库（比如InfluxDB）配合关系型数据库。时序库存原始事件，关系库存业务对象。别把所有东西塞进一张大表，查询会慢到你怀疑人生。

下面是一个简化的存储模型，我用的是Python + SQLAlchemy：

from sqlalchemy import Column, String, BigInteger, Float, DateTime, Enum
from sqlalchemy.ext.declarative import declarative_base
import enum

Base = declarative_base()

class OrderStatus(enum.Enum):
    NEW = 'NEW'
    PARTIALLY_FILLED = 'PARTIALLY_FILLED'
    FILLED = 'FILLED'
    CANCELED = 'CANCELED'
    REJECTED = 'REJECTED'

class TradeEvent(Base):
    __tablename__ = 'trade_events'
    
    event_id = Column(BigInteger, primary_key=True, autoincrement=True)
    event_type = Column(String(32), nullable=False)  # 'NEW_ORDER', 'FILL', 'CANCEL'
    symbol = Column(String(16), nullable=False)
    order_id = Column(String(64), nullable=False, index=True)
    price = Column(Float)
    quantity = Column(Float)
    status = Column(Enum(OrderStatus))
    timestamp = Column(DateTime, nullable=False, index=True)
    raw_payload = Column(String(2048))  # 原始JSON，用于审计

class AuditSnapshot(Base):
    __tablename__ = 'audit_snapshots'
    
    snapshot_id = Column(BigInteger, primary_key=True)
    snapshot_time = Column(DateTime, nullable=False, index=True)
    positions_json = Column(String(8192))  # 全量持仓序列化
    orders_json = Column(String(8192))     # 全量活动订单序列化
    checksum = Column(String(64))          # 用于校验完整性

你可能会问：「存原始JSON有什么用？」我告诉你，太有用了。有一次监管问我们某个订单的报价时间戳和交易所的时间戳差了2毫秒，我们直接把原始消息翻出来，发现是网络延迟，不是系统作弊。没有原始数据，你百口莫辩。

15.2 监管报告生成：MiFID II 和 Dodd-Frank 的「噩梦」

做市商如果涉及欧美市场，MiFID II（欧洲）和 Dodd-Frank（美国）是绕不开的两座大山。这两个法规的核心要求是什么？说白了就是：你做的每一笔交易，都要在指定时间内，按指定格式，报告给监管机构。

MiFID II 要求交易报告在 T+1 日内提交，包含65个字段。Dodd-Frank 的 swap 报告更狠，要求实时或准实时报告。我刚开始接触时，觉得这简直是反人类——65个字段，每个字段都有严格的格式要求，填错一个就罚款。

举个例子，MiFID II 里有个字段叫「交易对手识别码」（LEI），必须是20位字母数字。如果你填了19位或者21位，系统直接拒收。我曾经因为一个字段的时区格式写错了（UTC vs UTC+1），被退回了一整批报告，加班到凌晨三点重新生成。

避坑指南：千万不要手动生成报告。一定要用自动化管道。我见过有人用Excel手工填，结果漏了一笔交易，罚款5万欧元。自动化管道至少能保证「不漏、不错、不迟」。

下面是一个简化的报告生成流水线架构：

# 伪代码：MiFID II 报告生成器
class MiFIDReportGenerator:
    def __init__(self, trade_store, report_client):
        self.trade_store = trade_store
        self.report_client = report_client
    
    def generate_daily_report(self, date):
        # 1. 获取当天所有成交
        trades = self.trade_store.get_trades_by_date(date)
        
        # 2. 转换为MiFID II格式
        report_entries = []
        for trade in trades:
            entry = {
                'reporting_firm_id': 'LEI1234567890',
                'trade_date': trade.exec_time.strftime('%Y-%m-%d'),
                'trade_time': trade.exec_time.strftime('%H:%M:%S.%f')[:-3],
                'instrument_id': trade.symbol,
                'price': round(trade.price, 4),
                'quantity': trade.quantity,
                'counterparty_id': trade.counterparty_lei,
                'transaction_type': 'BUY' if trade.side == 'BUY' else 'SELL',
                # ... 还有60多个字段
            }
            report_entries.append(entry)
        
        # 3. 校验字段
        errors = self.validate_entries(report_entries)
        if errors:
            raise ValueError(f"报告校验失败: {errors}")
        
        # 4. 提交到监管平台
        response = self.report_client.submit(report_entries)
        return response

你可能会觉得，65个字段写起来很烦。但更烦的是，每个字段的规则还在变。MiFID II 在2023年更新过一次，把「交易时间」的精度从毫秒改到了微秒。如果你的系统不支持微秒，那就等着被罚吧。

15.3 反洗钱（AML）检查：别让做市商变成洗钱通道

反洗钱，听起来像是银行的事。但做市商一样有责任。监管要求做市商对每一笔交易进行AML筛查，识别可疑交易模式。说白了，就是看有没有人利用你的流动性来洗钱。

常见的AML检查包括：

大额交易监控：单笔超过某个阈值（比如1万美元）的交易，需要标记并上报。
频繁小额交易：有人故意把大额拆成小额，规避阈值检查。这叫「结构化交易」（Structuring）。
异常交易模式：比如一个账户在短时间内大量买卖同一品种，且价格明显偏离市场。这可能是「对倒交易」（Wash Trading）。
制裁名单筛查：交易对手是否在OFAC（美国财政部外国资产控制办公室）的制裁名单上。

我在项目中遇到过最头疼的是「误报」问题。AML系统太敏感，一天能报几百个警报，但99%都是误报。合规团队的人天天看警报，看到最后都麻木了。真正有问题的交易反而被漏掉了。

我的经验：不要只依赖规则引擎。规则引擎（比如「单笔超过1万就报警」）太死板。引入机器学习模型，根据历史数据训练一个异常检测模型，可以大幅降低误报率。我们当时用了一个简单的孤立森林（Isolation Forest），误报率从90%降到了30%。

下面是一个简单的AML检查流程：

class AMLChecker:
    def __init__(self, sanction_list, threshold=10000):
        self.sanction_list = sanction_list
        self.threshold = threshold
    
    def check_trade(self, trade):
        alerts = []
        
        # 1. 制裁名单检查
        if trade.counterparty_id in self.sanction_list:
            alerts.append(f"交易对手在制裁名单中: {trade.counterparty_id}")
        
        # 2. 大额交易检查
        if trade.notional_value > self.threshold:
            alerts.append(f"大额交易: {trade.notional_value} > {self.threshold}")
        
        # 3. 结构化交易检查（简化版）
        recent_trades = self.get_recent_trades(trade.counterparty_id, minutes=5)
        total_value = sum(t.notional_value for t in recent_trades) + trade.notional_value
        if total_value > self.threshold and all(t.notional_value < self.threshold for t in recent_trades):
            alerts.append(f"疑似结构化交易: 累计 {total_value}")
        
        return alerts

嗯，这里要注意：AML检查不能只做一次。交易执行前要做一次预检查，执行后还要做一次后检查。因为有些信息（比如交易对手的最终受益人）可能在交易执行后才更新。

15.4 知识体系总览

说了这么多，我画了一张图，帮你把合规与审计的核心逻辑串起来。你看一眼就明白了：

这张图其实就讲了三件事：存好数据、生成报告、检查异常。三者缺一不可。数据存不好，报告就出不来；报告出不来，监管就找你麻烦；AML没做好，可能直接吊销牌照。

最后说一句，合规系统最好在交易系统设计之初就考虑进去。不要等到上线了再补，那就像房子盖好了再挖地基，成本高得吓人。我个人习惯在每一个交易接口后面都加一个「审计钩子」，所有数据自动流入合规模块，不需要人工干预。

好了，这一章就到这里。合规虽然枯燥，但它是做市商业务的「护城河」。没有它，你的策略再赚钱，也只是一座沙堡。

📚 课程目录

15、合规与审计：交易记录存储，监管报告生成（MiFID II/Dodd-Frank），反洗钱（AML）检查

15.1 交易记录存储：不只是写日志那么简单

15.2 监管报告生成：MiFID II 和 Dodd-Frank 的「噩梦」

15.3 反洗钱（AML）检查：别让做市商变成洗钱通道

15.4 知识体系总览