-- Every account's debits and credits must net to its balance
SELECT l.account_id,
       SUM(CASE WHEN l.direction = 'credit' THEN l.amount ELSE -l.amount END) AS computed,
       b.balance AS reported
FROM ledger_entries l
JOIN account_balances b ON b.account_id = l.account_id
GROUP BY l.account_id, b.balance
HAVING SUM(CASE WHEN l.direction = 'credit' THEN l.amount ELSE -l.amount END) <> b.balance;
-- Any rows returned = a reconciliation break to investigate

from decimal import Decimal, ROUND_HALF_UP

# Never use float for money: 0.1 + 0.2 != 0.3 in binary floating point
print(0.1 + 0.2)                          # 0.30000000000000004
print(Decimal("0.1") + Decimal("0.2"))    # 0.3

fee = (Decimal("1999.00") * Decimal("0.018")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
print(fee)   # 35.98

-- PostgreSQL: safe to re-run the same batch
INSERT INTO payments (event_id, account_id, amount_minor, status, batch_id, loaded_at)
SELECT event_id, account_id, amount_minor, status, :batch_id, NOW()
FROM staging_payments
ON CONFLICT (event_id) DO UPDATE
  SET status    = EXCLUDED.status,
      batch_id  = EXCLUDED.batch_id,
      loaded_at = EXCLUDED.loaded_at;

import pandas as pd

ours = pd.read_csv("our_settled.csv")         # event_id, amount_minor
theirs = pd.read_csv("partner_settlement.csv")  # event_id, amount_minor

m = ours.merge(theirs, on="event_id", how="outer",
               suffixes=("_ours", "_theirs"), indicator=True)
missing = m[m["_merge"] != "both"]
mismatch = m[(m["_merge"] == "both") & (m["amount_minor_ours"] != m["amount_minor_theirs"])]
print(len(missing), "missing,", len(mismatch), "amount mismatches")

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("velocity").getOrCreate()
events = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "payments").load()
    .selectExpr("CAST(value AS STRING) AS json")
    .select(F.from_json("json", "account_id STRING, amount_minor LONG, event_time TIMESTAMP").alias("e"))
    .select("e.*"))

velocity = (events.withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "5 minutes"), "account_id")
    .agg(F.count("*").alias("txn_count")))

velocity.writeStream.outputMode("update").format("console").start().awaitTermination()

import hashlib

def mask_card(number: str) -> str:
    digits = number.replace(" ", "")
    return "*" * (len(digits) - 4) + digits[-4:]

def pseudonymise(value: str, salt: str) -> str:
    return hashlib.sha256((salt + value).encode()).hexdigest()

print(mask_card("4111 1111 1111 1111"))   # ************1111
# Keep the salt in a secrets manager, never in code

Fintech practice project (synthetic data)
1. Generate payment events with duplicates and late arrivals
2. Load with idempotent upserts on event_id
3. Build a ledger table and daily balance check
4. Reconcile against a mock settlement file
5. Add audit columns and a short lineage diagram

Fintech DE interview prep
☐ Why never store money as float
☐ How my pipeline handles retries (idempotency)
☐ How I reconcile two sources
☐ Event time vs processing time
☐ How I protect sensitive fields

Skill map
Core:    SQL, Python, pipelines, Git
Fintech: reconciliation, idempotency, ledgers, audit columns
Stream:  Kafka basics, windowing, watermarks
Secure:  masking, access control, secrets management