Cut Transaction Reporting Time from 8 Hours to 45 Minutes with Automation

# Personal note: Learned this after corrupting 3 months of reports with timezone issues
import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime, timezone

class TransactionPipeline:
    def __init__(self, db_connection_string):
        self.engine = create_engine(db_connection_string)
        
    def extract_transactions(self, start_date, end_date):
        """Pull transactions with UTC normalization"""
        query = f"""
        SELECT 
            transaction_id,
            account_id,
            security_symbol,
            transaction_type,
            quantity,
            price,
            TIMESTAMP AT TIME ZONE 'UTC' as execution_time,
            settlement_date,
            counterparty_id,
            venue_code
        FROM transactions
        WHERE execution_time >= '{start_date}'
          AND execution_time < '{end_date}'
          AND status = 'SETTLED'
        """
        
        # Watch out: Always filter SETTLED status or you'll report cancelled trades
        return pd.read_sql(query, self.engine)

# Initialize with your connection
pipeline = TransactionPipeline('postgresql://user:pass@localhost:5432/trading_db')
transactions = pipeline.extract_transactions('2025-10-01', '2025-11-01')

print(f"Extracted {len(transactions):,} transactions")

from typing import Dict, List
import re

class RegulatoryValidator:
    # Personal note: These regex patterns came from 40 pages of SEC documentation
    
    SYMBOL_PATTERN = re.compile(r'^[A-Z]{1,5}$')  # Standard stock symbols
    CUSIP_PATTERN = re.compile(r'^[0-9]{3}[0-9A-Z]{6}[0-9]$')
    
    def __init__(self):
        self.errors = []
        
    def validate_sec_606(self, df: pd.DataFrame) -> Dict:
        """Validate against SEC Rule 606 requirements"""
        required_fields = [
            'security_symbol', 'transaction_type', 
            'execution_time', 'venue_code'
        ]
        
        # Check for missing required fields
        missing = [f for f in required_fields if f not in df.columns]
        if missing:
            self.errors.append(f"Missing required fields: {missing}")
            
        # Validate security symbols
        invalid_symbols = df[~df['security_symbol'].str.match(
            self.SYMBOL_PATTERN, na=False
        )]
        
        if len(invalid_symbols) > 0:
            self.errors.append(
                f"Found {len(invalid_symbols)} invalid symbols"
            )
            
        # Check execution timestamps are in market hours (9:30 AM - 4:00 PM ET)
        df['exec_hour'] = pd.to_datetime(df['execution_time']).dt.hour
        after_hours = df[(df['exec_hour'] < 9) | (df['exec_hour'] >= 16)]
        
        return {
            'valid': len(self.errors) == 0,
            'errors': self.errors,
            'after_hours_trades': len(after_hours),
            'total_validated': len(df)
        }

validator = RegulatoryValidator()
validation_result = validator.validate_sec_606(transactions)

if validation_result['valid']:
    print("✓ All transactions passed validation")
else:
    print(f"✗ Validation failed: {validation_result['errors']}")

from jinja2 import Template
import csv

class ReportGenerator:
    def __init__(self):
        self.report_templates = {
            'sec_606': self._load_sec_template(),
            'finra_oats': self._load_finra_template()
        }
        
    def generate_sec_606_report(self, df: pd.DataFrame, month: str) -> str:
        """Generate SEC Rule 606 quarterly disclosure"""
        
        # Aggregate by venue and order type
        venue_stats = df.groupby(['venue_code', 'transaction_type']).agg({
            'transaction_id': 'count',
            'price': 'mean',
            'quantity': 'sum'
        }).reset_index()
        
        venue_stats.columns = [
            'Venue', 'Order_Type', 'Total_Orders', 
            'Avg_Price', 'Total_Shares'
        ]
        
        # Calculate payment for order flow (PFOF) if applicable
        venue_stats['PFOF_Amount'] = venue_stats.apply(
            lambda x: self._calculate_pfof(x['Venue'], x['Total_Shares']), 
            axis=1
        )
        
        # Generate report file
        filename = f'SEC_606_Report_{month}.csv'
        venue_stats.to_csv(filename, index=False, quoting=csv.QUOTE_ALL)
        
        return filename
    
    def _calculate_pfof(self, venue: str, shares: int) -> float:
        """Calculate payment for order flow based on venue agreements"""
        # Personal note: These rates change quarterly, pull from config
        pfof_rates = {
            'NASDAQ': 0.0018,
            'NYSE': 0.0015,
            'CBOE': 0.0020
        }
        return shares * pfof_rates.get(venue, 0.0)

generator = ReportGenerator()
report_file = generator.generate_sec_606_report(transactions, '2025-10')

print(f"Report generated: {report_file}")
print(f"Size: {os.path.getsize(report_file) / 1024:.1f} KB")

import schedule
import time
import smtplib
from email.mime.text import MIMEText

class ReportScheduler:
    def __init__(self, notification_email: str):
        self.email = notification_email
        
    def monthly_reporting_job(self):
        """Run on the 5th of each month"""
        try:
            # Run the full pipeline
            start = datetime.now()
            
            pipeline = TransactionPipeline(DB_CONNECTION)
            last_month = (datetime.now().replace(day=1) - 
                         timedelta(days=1))
            
            transactions = pipeline.extract_transactions(
                last_month.replace(day=1),
                datetime.now().replace(day=1)
            )
            
            validator = RegulatoryValidator()
            validation = validator.validate_sec_606(transactions)
            
            if not validation['valid']:
                self._send_alert(f"Validation failed: {validation['errors']}")
                return
            
            generator = ReportGenerator()
            report = generator.generate_sec_606_report(
                transactions, 
                last_month.strftime('%Y-%m')
            )
            
            elapsed = (datetime.now() - start).total_seconds() / 60
            self._send_success(report, elapsed)
            
        except Exception as e:
            self._send_alert(f"Pipeline failed: {str(e)}")
    
    def _send_success(self, report_file: str, minutes: float):
        """Send success notification"""
        msg = MIMEText(
            f"Monthly report generated successfully\n"
            f"File: {report_file}\n"
            f"Processing time: {minutes:.1f} minutes"
        )
        msg['Subject'] = '✓ Regulatory Report Complete'
        msg['To'] = self.email
        self._send_email(msg)

# Schedule to run monthly
scheduler = ReportScheduler('compliance@yourcompany.com')
schedule.every().month.at("05:00").do(scheduler.monthly_reporting_job)

# Keep the scheduler running
while True:
    schedule.run_pending()
    time.sleep(3600)  # Check every hour