
At Skillshare, I led the payments team in building a revenue recognition pipeline that accurately tracked revenue according to accounting standards. The system processed millions of transactions and generated financial reports.

#### The Challenge

Revenue recognition requires:
- Accurate timing (when to recognize revenue)
- Proper categorization (subscription vs one-time)
- Compliance with accounting standards (ASC 606)
- Audit trails for all calculations
- Reporting for finance team

#### Architecture

We built:
- **Data pipeline**: Extract transactions from payment processors
- **Recognition engine**: Apply recognition rules
- **Storage**: PostgreSQL for recognized revenue
- **Reporting**: Generate financial reports
- **Reconciliation**: Match with payment processor data

#### Recognition Rules

```python
from decimal import Decimal
from datetime import datetime, timedelta
from enum import Enum

class RecognitionMethod(Enum):
    IMMEDIATE = "immediate"
    DEFERRED = "deferred"
    SUBSCRIPTION = "subscription"

def recognize_revenue(
    transaction: Transaction,
    method: RecognitionMethod
) -> List[RevenueRecognition]:
    """
    Recognize revenue based on method.
    """
    if method == RecognitionMethod.IMMEDIATE:
        return [RevenueRecognition(
            amount=transaction.amount,
            date=transaction.date,
            transaction_id=transaction.id,
        )]
    
    elif method == RecognitionMethod.SUBSCRIPTION:
        # Recognize over subscription period
        period_start = transaction.date
        period_end = transaction.subscription_end_date
        days = (period_end - period_start).days
        
        recognitions = []
        daily_amount = transaction.amount / Decimal(str(days))
        
        current_date = period_start
        while current_date <= period_end:
            recognitions.append(RevenueRecognition(
                amount=daily_amount,
                date=current_date,
                transaction_id=transaction.id,
            ))
            current_date += timedelta(days=1)
        
        return recognitions
```

#### Data Pipeline

```python
import pandas as pd

def process_revenue_recognition(date: datetime) -> pd.DataFrame:
    """
    Process revenue recognition for a specific date.
    """
    # Fetch transactions
    transactions = fetch_transactions(date)
    
    # Apply recognition rules
    recognitions = []
    for transaction in transactions:
        method = determine_recognition_method(transaction)
        revenue = recognize_revenue(transaction, method)
        recognitions.extend(revenue)
    
    # Aggregate by date
    df = pd.DataFrame(recognitions)
    daily_revenue = df.groupby('date')['amount'].sum().reset_index()
    
    # Store in database
    store_revenue_recognition(daily_revenue)
    
    return daily_revenue
```

#### Reporting

```python
def generate_revenue_report(start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """
    Generate revenue report for date range.
    """
    query = """
        SELECT 
            date,
            SUM(amount) as recognized_revenue,
            COUNT(DISTINCT transaction_id) as transaction_count
        FROM revenue_recognitions
        WHERE date BETWEEN %s AND %s
        GROUP BY date
        ORDER BY date
    """
    
    df = pd.read_sql(query, conn, params=[start_date, end_date])
    return df
```

#### Reconciliation

```python
def reconcile_revenue(
    recognized_revenue: Decimal,
    payment_processor_revenue: Decimal,
    date: datetime
) -> ReconciliationResult:
    """
    Reconcile recognized revenue with payment processor data.
    """
    difference = abs(recognized_revenue - payment_processor_revenue)
    threshold = Decimal('0.01')  # Allow small rounding differences
    
    if difference > threshold:
        return ReconciliationResult(
            status='discrepancy',
            difference=difference,
            date=date,
        )
    
    return ReconciliationResult(
        status='reconciled',
        difference=difference,
        date=date,
    )
```

#### Testing

```python
def test_subscription_revenue_recognition():
    transaction = Transaction(
        id='t1',
        amount=Decimal('100.00'),
        date=datetime(2023, 1, 1),
        subscription_end_date=datetime(2023, 1, 31),
    )
    
    recognitions = recognize_revenue(transaction, RecognitionMethod.SUBSCRIPTION)
    
    assert len(recognitions) == 31  # 31 days
    assert sum(r.amount for r in recognitions) == Decimal('100.00')
```

#### Results

- Accurate revenue recognition
- Compliance with accounting standards
- Automated reporting
- Better financial visibility
- Reduced manual effort

> "Revenue recognition requires precision and compliance."

#### Lessons Learned

1. Understand accounting standards thoroughly
2. Test with real transaction data
3. Maintain detailed audit trails
4. Reconcile regularly
5. Work closely with finance team



Building a revenue recognition pipeline in Python


Testing financial software requires extra care-bugs can cost money and violate compliance. After building payment systems at Mercury and Heard, I've learned that testing strategies need to balance thoroughness with practicality.

#### The Challenge

Financial code must be:
- Accurate (no calculation errors)
- Secure (no vulnerabilities)
- Compliant (meets regulations)
- Reliable (handles edge cases)

#### Testing Pyramid

**Unit Tests (70%)**
Test individual functions and classes in isolation.

**Integration Tests (20%)**
Test how components work together.

**End-to-End Tests (10%)**
Test complete user flows.

#### Example: Testing Payment Processing

```python
import pytest
from decimal import Decimal
from payment_processor import process_payment

def test_process_payment_success():
    result = process_payment(
        amount=Decimal('100.50'),
        currency='USD',
        recipient_id='user123'
    )
    assert result.success is True
    assert result.transaction_id is not None

def test_process_payment_insufficient_funds():
    with pytest.raises(InsufficientFundsError):
        process_payment(
            amount=Decimal('1000.00'),
            currency='USD',
            recipient_id='user123'
        )

def test_process_payment_negative_amount():
    with pytest.raises(ValueError):
        process_payment(
            amount=Decimal('-100.00'),
            currency='USD',
            recipient_id='user123'
        )
```

#### Best Practices

1. Use Decimal for money (never float)
2. Test edge cases (zero, negative, very large amounts)
3. Mock external APIs (banks, payment processors)
4. Test idempotency (retries don't cause issues)
5. Test concurrency (race conditions)

> "In fintech, a bug isn't just a bug-it's a potential financial loss."

#### Tools

- pytest for testing framework
- pytest-asyncio for async tests
- pytest-mock for mocking
- hypothesis for property-based testing
- coverage.py for code coverage



Python testing strategies for fintech


At Mercury, we needed to process millions of transactions daily for reconciliation, reporting, and compliance. Python's data ecosystem-Pandas, Polars, and Dask-made this manageable, but building reliable pipelines required careful design.

#### The Challenge

Financial data pipelines must be:
- Accurate (no data loss or corruption)
- Fast (process millions of records)
- Reliable (handle failures gracefully)
- Auditable (track every transformation)

#### Our Architecture

We used a combination of:
- **Pandas** for in-memory transformations
- **Polars** for larger datasets (faster, more memory-efficient)
- **Dask** for distributed processing
- **Apache Airflow** for orchestration

#### Example Pipeline

```python
import pandas as pd
import polars as pl
from dask import dataframe as dd

def process_transactions(file_path: str) -> pd.DataFrame:
    # Read with Polars for speed
    df = pl.read_csv(file_path)
    
    # Transformations
    df = df.filter(pl.col("amount") > 0)
    df = df.with_columns([
        pl.col("timestamp").str.strptime(pl.Datetime),
        pl.col("amount").cast(pl.Decimal(10, 2))
    ])
    
    # Convert to Pandas for final processing
    return df.to_pandas()

def reconcile_transactions(transactions: pd.DataFrame, 
                          ledger: pd.DataFrame) -> pd.DataFrame:
    merged = transactions.merge(
        ledger, 
        on=["transaction_id", "amount"],
        how="outer",
        indicator=True
    )
    
    # Find discrepancies
    discrepancies = merged[merged["_merge"] != "both"]
    return discrepancies
```

#### Error Handling

```python
def safe_process(file_path: str) -> Optional[pd.DataFrame]:
    try:
        return process_transactions(file_path)
    except FileNotFoundError:
        logger.error(f"File not found: {file_path}")
        return None
    except pd.errors.EmptyDataError:
        logger.warning(f"Empty file: {file_path}")
        return pd.DataFrame()
    except Exception as e:
        logger.exception(f"Unexpected error: {e}")
        raise
```

#### Performance Optimization

- Use Polars for large datasets
- Parallelize with Dask when possible
- Cache intermediate results
- Use appropriate data types
- Profile and optimize bottlenecks

#### Testing

```python
def test_transaction_processing():
    test_data = pd.DataFrame({
        "transaction_id": ["T1", "T2"],
        "amount": [100.50, 200.75],
        "timestamp": ["2023-04-01 10:00:00", "2023-04-01 11:00:00"]
    })
    
    result = process_transactions(test_data)
    assert len(result) == 2
    assert result["amount"].sum() == 301.25
```

#### Lessons Learned

1. Start with Pandas, optimize with Polars when needed
2. Always validate data at pipeline boundaries
3. Log everything for debugging
4. Test with production-like data volumes
5. Monitor pipeline performance and costs

> "Data pipelines are only as good as their error handling."



Building data pipelines for fintech with Python


Processing payments at scale requires handling thousands of concurrent requests efficiently. Python's async/await syntax, combined with FastAPI, provides an excellent foundation for building high-performance payment APIs.

#### Why Async Matters

Synchronous payment processing blocks threads while waiting for external APIs (banks, payment processors, fraud checks). With async, we can handle many requests concurrently without blocking.

#### The Pattern

```python
from fastapi import FastAPI
import httpx

app = FastAPI()

async def process_payment(payment_id: str):
    async with httpx.AsyncClient() as client:
        # These run concurrently
        fraud_check = await client.post("/fraud/check", json=payment_data)
        bank_validation = await client.post("/bank/validate", json=payment_data)
        
        if fraud_check.status_code == 200 and bank_validation.status_code == 200:
            return await execute_payment(payment_id)
```

#### Real-World Performance

At Mercury, moving to async patterns increased our payment processing throughput by 3x. The key was understanding when to use async (I/O-bound operations) versus when to use sync (CPU-bound calculations).

#### Gotchas

- Database connection pooling is critical
- Error handling becomes more complex
- Testing async code requires different patterns
- Monitoring needs to account for concurrent operations

#### Best Practices

1. Use async for I/O-bound operations
2. Keep CPU-bound work synchronous or use process pools
3. Implement proper connection pooling
4. Add comprehensive error handling and retries
5. Monitor concurrency metrics



Python async patterns for payment processing


After years of experience in engineering leadership and fintech, I've learned that python async best practices requires careful attention to both technical excellence and human factors.

#### The Challenge

Python async best practices is something I've encountered throughout my career at companies like Mercury, Heard, Ness, and now Pangea. The key is understanding both the technical and human aspects.

#### Key Insights

1. **Technical Excellence**: Whether it's Python, TypeScript, or infrastructure, solid technical foundations matter.

2. **People First**: Engineering is ultimately about people-building systems that serve users and teams that thrive.

3. **Balance**: Finding the right balance between technical depth and practical application.

#### Lessons Learned

- Start with understanding the problem deeply
- Build iteratively and get feedback early
- Invest in both technical skills and soft skills
- Take care of your mental and physical health

> "The best engineering solutions come from understanding both the code and the people who use it."

#### Next Steps

Consider how python async best practices applies to your own work. What can you learn? What can you improve? How can you help others?



Python async best practices


After years of experience in engineering leadership and fintech, I've learned that payment reconciliation automation requires careful attention to both technical excellence and human factors.

#### The Challenge

Payment reconciliation automation is something I've encountered throughout my career at companies like Mercury, Heard, Ness, and now Pangea. The key is understanding both the technical and human aspects.

#### Key Insights

1. **Technical Excellence**: Whether it's Python, TypeScript, or infrastructure, solid technical foundations matter.

2. **People First**: Engineering is ultimately about people-building systems that serve users and teams that thrive.

3. **Balance**: Finding the right balance between technical depth and practical application.

#### Lessons Learned

- Start with understanding the problem deeply
- Build iteratively and get feedback early
- Invest in both technical skills and soft skills
- Take care of your mental and physical health

> "The best engineering solutions come from understanding both the code and the people who use it."

#### Next Steps

Consider how payment reconciliation automation applies to your own work. What can you learn? What can you improve? How can you help others?



Payment reconciliation automation


After years of experience in engineering leadership and fintech, I've learned that fx market data pipelines requires careful attention to both technical excellence and human factors.

#### The Challenge

FX market data pipelines is something I've encountered throughout my career at companies like Mercury, Heard, Ness, and now Pangea. The key is understanding both the technical and human aspects.

#### Key Insights

1. **Technical Excellence**: Whether it's Python, TypeScript, or infrastructure, solid technical foundations matter.

2. **People First**: Engineering is ultimately about people-building systems that serve users and teams that thrive.

3. **Balance**: Finding the right balance between technical depth and practical application.

#### Lessons Learned

- Start with understanding the problem deeply
- Build iteratively and get feedback early
- Invest in both technical skills and soft skills
- Take care of your mental and physical health

> "The best engineering solutions come from understanding both the code and the people who use it."

#### Next Steps

Consider how fx market data pipelines applies to your own work. What can you learn? What can you improve? How can you help others?



FX market data pipelines


After years of experience in engineering leadership and fintech, I've learned that building secure payment apis requires careful attention to both technical excellence and human factors.

#### The Challenge

Building secure payment APIs is something I've encountered throughout my career at companies like Mercury, Heard, Ness, and now Pangea. The key is understanding both the technical and human aspects.

#### Key Insights

1. **Technical Excellence**: Whether it's Python, TypeScript, or infrastructure, solid technical foundations matter.

2. **People First**: Engineering is ultimately about people-building systems that serve users and teams that thrive.

3. **Balance**: Finding the right balance between technical depth and practical application.

#### Lessons Learned

- Start with understanding the problem deeply
- Build iteratively and get feedback early
- Invest in both technical skills and soft skills
- Take care of your mental and physical health

> "The best engineering solutions come from understanding both the code and the people who use it."

#### Next Steps

Consider how building secure payment apis applies to your own work. What can you learn? What can you improve? How can you help others?



Building secure payment APIs


At Pangea, we're building real-time FX trading systems that need to handle market data updates, execute trades, and maintain position state. Python's async capabilities combined with proper architecture make this possible.

#### The Challenge

FX trading requires:
- Low latency (milliseconds matter)
- High throughput (thousands of updates per second)
- Reliability (no data loss)
- Consistency (accurate position tracking)

#### Architecture

We use:
- **asyncio** for concurrent processing
- **WebSockets** for real-time market data
- **Redis** for position state
- **PostgreSQL** for persistence
- **Message queues** for order execution

#### Market Data Handler

```python
import asyncio
import websockets
import json

class FXMarketDataHandler:
    def __init__(self):
        self.subscribers = set()
        self.positions = {}
    
    async def handle_market_data(self, websocket, path):
        async for message in websocket:
            data = json.loads(message)
            await self.process_update(data)
            await self.notify_subscribers(data)
    
    async def process_update(self, data):
        # Update positions in Redis
        currency_pair = data['pair']
        rate = data['rate']
        # Process update...
```

#### Performance Considerations

- Use connection pooling
- Batch database writes
- Cache frequently accessed data
- Monitor latency metrics
- Optimize hot paths

#### Lessons Learned

1. Async is essential for real-time systems
2. State management is critical
3. Monitoring and alerting are non-negotiable
4. Test with production-like load
5. Plan for failure scenarios

> "In trading systems, milliseconds matter. But correctness matters more."



Building real-time FX trading systems


After a decade away from hands-on coding, diving back into Python for payment processing has been both challenging and invigorating. At Mercury, we needed robust pipelines for handling 1099 interest payouts, and Python's ecosystem proved perfect for the job.

#### Why Python for Payments?

Python's data processing libraries-Pandas, NumPy, and the async capabilities of FastAPI-make it ideal for financial data transformations. The type hints in modern Python also help catch errors before they hit production, which is critical when dealing with money.

#### The Pipeline Architecture

We built an event-driven system using Python's asyncio and message queues. Each payment goes through validation, transformation, and reconciliation stages. The key was making each stage idempotent and traceable.

> "When dealing with financial transactions, every operation must be reversible and auditable."

#### Lessons Learned

The biggest challenge wasn't the code itself, but ensuring compliance and audit trails. Python's logging and monitoring libraries made this manageable, but it required discipline in how we structured the code.



Building payment pipelines with Python


At Mercury, we built an automated reconciliation system that reduced manual effort by 60%. The system matches transactions between our ledger and partner bank statements automatically, flagging discrepancies for review.

#### The Problem

Manual reconciliation was:
- Time-consuming (hours per day)
- Error-prone (human mistakes)
- Not scalable (more transactions = more time)
- Expensive (high labor costs)

#### The Solution

We built an automated system using:
- Python for data processing
- Pandas for matching logic
- PostgreSQL for transaction storage
- Scheduled jobs for daily reconciliation
- Alerting for discrepancies

#### Matching Algorithm

```python
import pandas as pd
from decimal import Decimal

def reconcile_transactions(
    ledger_transactions: pd.DataFrame,
    bank_statement: pd.DataFrame
) -> pd.DataFrame:
    """
    Match transactions between ledger and bank statement.
    """
    # Normalize amounts
    ledger_transactions['amount_abs'] = ledger_transactions['amount'].abs()
    bank_statement['amount_abs'] = bank_statement['amount'].abs()
    
    # Match by amount and date (within 2 days)
    matches = pd.merge_asof(
        ledger_transactions.sort_values('date'),
        bank_statement.sort_values('date'),
        left_on='date',
        right_on='date',
        by='amount_abs',
        tolerance=pd.Timedelta('2 days'),
        direction='nearest'
    )
    
    # Find unmatched transactions
    matched_ledger_ids = matches['ledger_id'].dropna()
    unmatched_ledger = ledger_transactions[
        ~ledger_transactions['id'].isin(matched_ledger_ids)
    ]
    
    matched_bank_ids = matches['bank_id'].dropna()
    unmatched_bank = bank_statement[
        ~bank_statement['id'].isin(matched_bank_ids)
    ]
    
    return {
        'matched': matches.dropna(),
        'unmatched_ledger': unmatched_ledger,
        'unmatched_bank': unmatched_bank,
    }
```

#### Handling Edge Cases

- **Timing differences**: Transactions posted on different days
- **Fees**: Bank fees not in our ledger
- **Reversals**: Failed transactions that were reversed
- **Duplicates**: Same transaction appearing twice
- **Rounding**: Small differences due to rounding

#### Automation

```python
from apscheduler.schedulers.blocking import BlockingScheduler

def daily_reconciliation():
    """Run reconciliation daily at 2 AM."""
    ledger = fetch_ledger_transactions()
    bank_statement = fetch_bank_statement()
    
    results = reconcile_transactions(ledger, bank_statement)
    
    # Alert on discrepancies
    if len(results['unmatched_ledger']) > 0:
        send_alert('Unmatched ledger transactions', results['unmatched_ledger'])
    
    if len(results['unmatched_bank']) > 0:
        send_alert('Unmatched bank transactions', results['unmatched_bank'])
    
    # Store results
    save_reconciliation_results(results)

scheduler = BlockingScheduler()
scheduler.add_job(daily_reconciliation, 'cron', hour=2)
scheduler.start()
```

#### Results

- 60% reduction in manual effort
- Faster reconciliation (minutes vs hours)
- Fewer errors
- Better audit trail
- Scalable to more transactions

> "Automation frees up time for higher-value work."

#### Best Practices

1. Start with simple matching rules
2. Iterate based on results
3. Handle edge cases explicitly
4. Log everything for debugging
5. Alert on discrepancies promptly



Automated reconciliation with partner banks


At Heard, we built an AI-powered auto-categorization system that reduced manual bookkeeping time by 40%. The system uses Python with machine learning libraries to categorize transactions automatically.

#### The Problem

Mental health professionals spend hours each week categorizing expenses for tax purposes. This is tedious, error-prone work that takes time away from their practice.

#### The Solution

We built a classification system using:
- Scikit-learn for initial models
- Transformers for better accuracy
- FastAPI for the API layer
- PostgreSQL for transaction storage

#### The Architecture

```python
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB

class TransactionCategorizer:
    def __init__(self):
        self.vectorizer = TfidfVectorizer()
        self.classifier = MultinomialNB()
    
    def train(self, transactions, categories):
        X = self.vectorizer.fit_transform([t.description for t in transactions])
        self.classifier.fit(X, categories)
    
    def predict(self, transaction):
        X = self.vectorizer.transform([transaction.description])
        return self.classifier.predict(X)[0]
```

#### Results

- 40% reduction in manual categorization time
- 92% accuracy on common categories
- Continuous learning from user corrections
- Significant user satisfaction improvement

#### Lessons Learned

1. Start with simple models, iterate based on feedback
2. User corrections are valuable training data
3. Explainability matters-users need to trust the system
4. Handle edge cases gracefully

> "AI should augment human judgment, not replace it."



AI-powered auto-categorization in accounting software


At Mercury, we built systems to calculate and distribute 1099 interest payouts for thousands of business customers. Accurate tax reporting is critical-mistakes can result in penalties and customer complaints.

#### The Requirements

1099-INT forms require:
- Accurate interest calculations
- Proper tax year reporting
- Customer information (SSN/EIN)
- Compliance with IRS regulations
- Timely filing (by January 31)

#### Architecture

We built:
- **Calculation engine**: Python service to calculate interest
- **Data pipeline**: Extract, transform, load (ETL) for tax data
- **Form generation**: Create 1099-INT PDFs
- **Filing system**: Submit to IRS and states
- **Customer portal**: Allow customers to download forms

#### Interest Calculation

```python
from decimal import Decimal
from datetime import datetime, timedelta

def calculate_interest(
    principal: Decimal,
    annual_rate: Decimal,
    start_date: datetime,
    end_date: datetime
) -> Decimal:
    """
    Calculate interest using daily compounding.
    """
    days = (end_date - start_date).days
    daily_rate = annual_rate / Decimal('365')
    
    # Simple interest calculation
    interest = principal * daily_rate * Decimal(str(days))
    
    return interest.quantize(Decimal('0.01'))

def calculate_annual_interest(
    account_id: str,
    year: int
) -> Decimal:
    """
    Calculate total interest for a tax year.
    """
    transactions = get_interest_transactions(account_id, year)
    
    total_interest = Decimal('0')
    for transaction in transactions:
        if transaction.type == 'interest_payment':
            total_interest += transaction.amount
    
    return total_interest
```

#### Data Pipeline

```python
import pandas as pd

def prepare_1099_data(year: int) -> pd.DataFrame:
    """
    Prepare data for 1099-INT forms.
    """
    # Fetch all accounts with interest
    accounts = get_accounts_with_interest(year)
    
    data = []
    for account in accounts:
        interest = calculate_annual_interest(account.id, year)
        
        if interest > Decimal('10'):  # IRS threshold
            data.append({
                'account_id': account.id,
                'customer_name': account.customer_name,
                'tax_id': account.tax_id,  # SSN or EIN
                'interest_amount': float(interest),
                'year': year,
            })
    
    return pd.DataFrame(data)
```

#### Form Generation

```python
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas

def generate_1099_form(customer_data: dict) -> bytes:
    """
    Generate 1099-INT PDF form.
    """
    buffer = io.BytesIO()
    c = canvas.Canvas(buffer, pagesize=letter)
    
    # Form fields
    c.drawString(100, 700, f"PAYER'S name: {customer_data['payer_name']}")
    c.drawString(100, 680, f"RECIPIENT'S name: {customer_data['customer_name']}")
    c.drawString(100, 660, f"Tax ID: {customer_data['tax_id']}")
    c.drawString(100, 640, f"Interest income: ${customer_data['interest_amount']:.2f}")
    
    c.save()
    buffer.seek(0)
    return buffer.read()
```

#### Compliance

- Verify tax IDs (SSN/EIN format)
- Handle backup withholding
- Report to IRS and states
- Maintain records for 7 years
- Handle corrections and amendments

#### Testing

```python
def test_interest_calculation():
    principal = Decimal('10000')
    rate = Decimal('0.05')  # 5%
    start = datetime(2023, 1, 1)
    end = datetime(2023, 12, 31)
    
    interest = calculate_interest(principal, rate, start, end)
    expected = Decimal('500.00')
    
    assert abs(interest - expected) < Decimal('0.01')
```

#### Results

- Accurate tax reporting for thousands of customers
- Timely filing with IRS
- Reduced manual effort
- Better customer experience
- Compliance with regulations

> "Tax reporting requires precision and attention to detail."

#### Lessons Learned

1. Start early (don't wait until January)
2. Test thoroughly with sample data
3. Handle edge cases (closed accounts, corrections)
4. Communicate with customers proactively
5. Keep detailed audit trails

