Building Audit Trails for Financial Systems

1Events              Audit Log           Verification        Query Layer
2──────────────      ─────────────       ────────────────    ─────────────
3Order created  →    Append-only    →    Chain hash     →    PostgreSQL
4Order modified      Event stream        verification        Indexed by
5Order canceled                                              order_id,
6Trade executed                                              user_id, time
7

1from dataclasses import dataclass, field
2from typing import Dict, Any, Optional
3from datetime import datetime
4import hashlib
5import json
6import uuid
7
8@dataclass
9class AuditEvent:
10    """Immutable audit event."""
11    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
12    timestamp: datetime = field(default_factory=datetime.utcnow)
13    event_type: str = ""
14    entity_type: str = ""  # order, trade, position
15    entity_id: str = ""
16    user_id: str = ""
17    action: str = ""  # create, modify, cancel, execute
18    previous_state: Optional[Dict[str, Any]] = None
19    new_state: Dict[str, Any] = field(default_factory=dict)
20    metadata: Dict[str, Any] = field(default_factory=dict)
21    previous_event_hash: Optional[str] = None
22    event_hash: Optional[str] = None
23    
24    def calculate_hash(self) -> str:
25        """
26        Calculate cryptographic hash of event.
27        
28        Creates tamper-evident chain by including previous event hash.
29        """
30        data = {
31            'event_id': self.event_id,
32            'timestamp': self.timestamp.isoformat(),
33            'event_type': self.event_type,
34            'entity_type': self.entity_type,
35            'entity_id': self.entity_id,
36            'user_id': self.user_id,
37            'action': self.action,
38            'previous_state': self.previous_state,
39            'new_state': self.new_state,
40            'metadata': self.metadata,
41            'previous_event_hash': self.previous_event_hash,
42        }
43        
44        # Canonical JSON (sorted keys)
45        canonical = json.dumps(data, sort_keys=True)
46        
47        return hashlib.sha256(canonical.encode()).hexdigest()
48    
49    def finalize(self, previous_event_hash: Optional[str] = None):
50        """Finalize event with hash chain."""
51        self.previous_event_hash = previous_event_hash
52        self.event_hash = self.calculate_hash()
53
54class AuditLog:
55    """Append-only audit log with hash chain verification."""
56    
57    def __init__(self, db_connection):
58        self.db = db_connection
59        self.last_event_hash: Optional[str] = None
60        self._load_last_hash()
61    
62    def _load_last_hash(self):
63        """Load hash of most recent event."""
64        cursor = self.db.cursor()
65        cursor.execute("""
66            SELECT event_hash FROM audit_events
67            ORDER BY timestamp DESC, event_id DESC
68            LIMIT 1
69        """)
70        
71        row = cursor.fetchone()
72        if row:
73            self.last_event_hash = row[0]
74    
75    def append(self, event: AuditEvent) -> str:
76        """
77        Append event to audit log.
78        
79        Returns:
80            Event ID
81        """
82        # Finalize event with hash chain
83        event.finalize(previous_event_hash=self.last_event_hash)
84        
85        # Insert into database
86        cursor = self.db.cursor()
87        cursor.execute("""
88            INSERT INTO audit_events (
89                event_id, timestamp, event_type, entity_type, entity_id,
90                user_id, action, previous_state, new_state, metadata,
91                previous_event_hash, event_hash
92            ) VALUES (
93                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
94            )
95        """, (
96            event.event_id,
97            event.timestamp,
98            event.event_type,
99            event.entity_type,
100            event.entity_id,
101            event.user_id,
102            event.action,
103            json.dumps(event.previous_state) if event.previous_state else None,
104            json.dumps(event.new_state),
105            json.dumps(event.metadata),
106            event.previous_event_hash,
107            event.event_hash
108        ))
109        
110        self.db.commit()
111        
112        # Update last hash
113        self.last_event_hash = event.event_hash
114        
115        return event.event_id
116    
117    def verify_chain(self, start_event_id: Optional[str] = None,
118                    end_event_id: Optional[str] = None) -> bool:
119        """
120        Verify hash chain integrity.
121        
122        Args:
123            start_event_id: Start verification from this event
124            end_event_id: End verification at this event
125            
126        Returns:
127            True if chain is valid
128        """
129        cursor = self.db.cursor()
130        
131        query = """
132            SELECT event_id, timestamp, event_type, entity_type, entity_id,
133                   user_id, action, previous_state, new_state, metadata,
134                   previous_event_hash, event_hash
135            FROM audit_events
136            WHERE 1=1
137        """
138        params = []
139        
140        if start_event_id:
141            query += " AND event_id >= %s"
142            params.append(start_event_id)
143        
144        if end_event_id:
145            query += " AND event_id <= %s"
146            params.append(end_event_id)
147        
148        query += " ORDER BY timestamp ASC, event_id ASC"
149        
150        cursor.execute(query, params)
151        
152        previous_hash = None
153        
154        for row in cursor:
155            event = AuditEvent(
156                event_id=row[0],
157                timestamp=row[1],
158                event_type=row[2],
159                entity_type=row[3],
160                entity_id=row[4],
161                user_id=row[5],
162                action=row[6],
163                previous_state=json.loads(row[7]) if row[7] else None,
164                new_state=json.loads(row[8]),
165                metadata=json.loads(row[9]),
166                previous_event_hash=row[10],
167                event_hash=row[11]
168            )
169            
170            # Check previous hash matches
171            if event.previous_event_hash != previous_hash:
172                print(f"Hash chain broken at event {event.event_id}")
173                print(f"Expected previous hash: {previous_hash}")
174                print(f"Actual previous hash: {event.previous_event_hash}")
175                return False
176            
177            # Recalculate and verify hash
178            calculated_hash = event.calculate_hash()
179            if calculated_hash != event.event_hash:
180                print(f"Event hash mismatch at {event.event_id}")
181                print(f"Stored hash: {event.event_hash}")
182                print(f"Calculated hash: {calculated_hash}")
183                return False
184            
185            previous_hash = event.event_hash
186        
187        return True
188
189

1-- Audit events table
2CREATE TABLE audit_events (
3    event_id UUID PRIMARY KEY,
4    timestamp TIMESTAMPTZ NOT NULL,
5    event_type VARCHAR(100) NOT NULL,
6    entity_type VARCHAR(50) NOT NULL,
7    entity_id VARCHAR(255) NOT NULL,
8    user_id VARCHAR(255) NOT NULL,
9    action VARCHAR(50) NOT NULL,
10    previous_state JSONB,
11    new_state JSONB NOT NULL,
12    metadata JSONB,
13    previous_event_hash VARCHAR(64),
14    event_hash VARCHAR(64) NOT NULL,
15    
16    -- Ensure append-only (no updates/deletes)
17    CONSTRAINT no_updates CHECK (false)
18);
19
20-- Prevent updates and deletes
21CREATE RULE audit_events_no_update AS
22    ON UPDATE TO audit_events
23    DO INSTEAD NOTHING;
24
25CREATE RULE audit_events_no_delete AS
26    ON DELETE TO audit_events
27    DO INSTEAD NOTHING;
28
29-- Indexes for common queries
30CREATE INDEX idx_audit_entity ON audit_events(entity_type, entity_id, timestamp DESC);
31CREATE INDEX idx_audit_user ON audit_events(user_id, timestamp DESC);
32CREATE INDEX idx_audit_timestamp ON audit_events(timestamp DESC);
33CREATE INDEX idx_audit_type ON audit_events(event_type, timestamp DESC);
34
35-- GIN index for metadata queries
36CREATE INDEX idx_audit_metadata ON audit_events USING GIN(metadata);
37
38-- Partitioning by month for performance
39CREATE TABLE audit_events_2024_01 PARTITION OF audit_events
40    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
41
42CREATE TABLE audit_events_2024_02 PARTITION OF audit_events
43    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
44-- ... create partitions for 7 years
45

1from enum import Enum
2
3class OrderAction(Enum):
4    CREATE = "create"
5    MODIFY = "modify"
6    CANCEL = "cancel"
7    PARTIAL_FILL = "partial_fill"
8    FILL = "fill"
9    REJECT = "reject"
10
11class OrderAuditTracker:
12    """Track complete order lifecycle for compliance."""
13    
14    def __init__(self, audit_log: AuditLog):
15        self.audit_log = audit_log
16    
17    def log_order_created(self, order_id: str, user_id: str, order_details: Dict[str, Any]):
18        """Log order creation."""
19        event = AuditEvent(
20            event_type="order_lifecycle",
21            entity_type="order",
22            entity_id=order_id,
23            user_id=user_id,
24            action=OrderAction.CREATE.value,
25            previous_state=None,
26            new_state=order_details,
27            metadata={
28                'source': 'trading_system',
29                'order_entry_time': datetime.utcnow().isoformat(),
30                'client_order_id': order_details.get('client_order_id'),
31            }
32        )
33        
34        return self.audit_log.append(event)
35    
36    def log_order_modified(self, order_id: str, user_id: str,
37                          previous_state: Dict[str, Any],
38                          new_state: Dict[str, Any],
39                          reason: str):
40        """Log order modification."""
41        event = AuditEvent(
42            event_type="order_lifecycle",
43            entity_type="order",
44            entity_id=order_id,
45            user_id=user_id,
46            action=OrderAction.MODIFY.value,
47            previous_state=previous_state,
48            new_state=new_state,
49            metadata={
50                'reason': reason,
51                'modification_time': datetime.utcnow().isoformat(),
52            }
53        )
54        
55        return self.audit_log.append(event)
56    
57    def log_order_filled(self, order_id: str, user_id: str,
58                        fill_details: Dict[str, Any]):
59        """Log order fill."""
60        event = AuditEvent(
61            event_type="order_lifecycle",
62            entity_type="order",
63            entity_id=order_id,
64            user_id=user_id,
65            action=OrderAction.FILL.value,
66            new_state=fill_details,
67            metadata={
68                'execution_time': datetime.utcnow().isoformat(),
69                'execution_venue': fill_details.get('exchange'),
70                'fill_price': fill_details.get('price'),
71                'fill_quantity': fill_details.get('quantity'),
72            }
73        )
74        
75        return self.audit_log.append(event)
76    
77    def get_order_history(self, order_id: str) -> List[AuditEvent]:
78        """Get complete audit trail for order."""
79        cursor = self.audit_log.db.cursor()
80        cursor.execute("""
81            SELECT event_id, timestamp, event_type, entity_type, entity_id,
82                   user_id, action, previous_state, new_state, metadata,
83                   previous_event_hash, event_hash
84            FROM audit_events
85            WHERE entity_type = 'order' AND entity_id = %s
86            ORDER BY timestamp ASC
87        """, (order_id,))
88        
89        events = []
90        for row in cursor:
91            events.append(AuditEvent(
92                event_id=row[0],
93                timestamp=row[1],
94                event_type=row[2],
95                entity_type=row[3],
96                entity_id=row[4],
97                user_id=row[5],
98                action=row[6],
99                previous_state=json.loads(row[7]) if row[7] else None,
100                new_state=json.loads(row[8]),
101                metadata=json.loads(row[9]),
102                previous_event_hash=row[10],
103                event_hash=row[11]
104            ))
105        
106        return events
107

1from typing import List, Tuple
2from datetime import date
3
4class RegulatoryReporter:
5    """Generate regulatory reports from audit trail."""
6    
7    def __init__(self, db_connection):
8        self.db = db_connection
9    
10    def trade_report_mifid_ii(self, start_date: date, end_date: date) -> List[Dict[str, Any]]:
11        """
12        Generate MiFID II trade report.
13        
14        Required fields:
15        - Trading venue
16        - Instrument identification
17        - Buy/sell indicator
18        - Quantity
19        - Price
20        - Trade date and time (UTC)
21        - Client identification
22        - Investment decision maker
23        - Execution decision maker
24        """
25        cursor = self.db.cursor()
26        cursor.execute("""
27            SELECT
28                event_id,
29                timestamp,
30                entity_id AS trade_id,
31                user_id,
32                new_state->>'symbol' AS instrument,
33                new_state->>'side' AS buy_sell_indicator,
34                (new_state->>'quantity')::numeric AS quantity,
35                (new_state->>'price')::numeric AS price,
36                new_state->>'exchange' AS trading_venue,
37                metadata->>'client_id' AS client_id,
38                metadata->>'investment_decision_maker' AS investment_decision_maker,
39                metadata->>'execution_decision_maker' AS execution_decision_maker
40            FROM audit_events
41            WHERE event_type = 'order_lifecycle'
42              AND action = 'fill'
43              AND timestamp::date BETWEEN %s AND %s
44            ORDER BY timestamp
45        """, (start_date, end_date))
46        
47        trades = []
48        for row in cursor:
49            trades.append({
50                'trade_id': row[0],
51                'trade_timestamp_utc': row[1].isoformat(),
52                'trader_id': row[2],
53                'instrument': row[3],
54                'buy_sell': row[4],
55                'quantity': float(row[5]),
56                'price': float(row[6]),
57                'venue': row[7],
58                'client_id': row[8],
59                'investment_decision': row[9],
60                'execution_decision': row[10],
61            })
62        
63        return trades
64    
65    def user_activity_report(self, user_id: str, start_date: date, end_date: date) -> List[Dict[str, Any]]:
66        """Generate user activity report for compliance review."""
67        cursor = self.db.cursor()
68        cursor.execute("""
69            SELECT
70                timestamp,
71                event_type,
72                action,
73                entity_type,
74                entity_id,
75                new_state,
76                metadata
77            FROM audit_events
78            WHERE user_id = %s
79              AND timestamp::date BETWEEN %s AND %s
80            ORDER BY timestamp
81        """, (user_id, start_date, end_date))
82        
83        activities = []
84        for row in cursor:
85            activities.append({
86                'timestamp': row[0].isoformat(),
87                'event_type': row[1],
88                'action': row[2],
89                'entity_type': row[3],
90                'entity_id': row[4],
91                'details': row[5],
92                'metadata': row[6],
93            })
94        
95        return activities
96

1Metric                          Value
2────────────────────────────────────────────────────
3Events logged                   2.4 billion
4Regulatory audits               6 (all passed)
5Hash chain verifications        12,500 (all valid)
6Avg write latency               0.8ms
7Avg query latency (order)       12ms
8Storage size                    850 GB (compressed)
9Oldest event                    2019-01-15
10

1Audit Type         Year    Duration    Findings    Result
2──────────────────────────────────────────────────────────────────
3MiFID II           2020    3 weeks     0           Pass
4Internal SOC2      2021    1 week      0           Pass
5MiFID II           2022    2 weeks     0           Pass
6Internal SOC2      2023    1 week      0           Pass
7MiFID II           2024    2 weeks     0           Pass
8