mrik8899/fintech-demo
0
1# app/simulator.py2# Live transaction simulator — generates realistic payment data in real-time.3# Background thread ticks every N seconds, advancing a simulation clock.4# Events (bank outage, merchant decline, amount anomaly, recovery) inject5# realistic disturbances that the pipeline detects.6 7import random8import threading9import time10import uuid11import os12from datetime import datetime, timedelta13from dataclasses import dataclass14from typing import Optional15 16from app.config import DB_PATH17 18# ── Constants (mirrored from seed.py — needed for transaction generation) ──19 20BANKS = ["HBL", "Bank Alfalah", "Meezan", "UBL", "HMB", "Bank of Punjab"]21 22CATEGORY_CONFIG = {23 "supermarket": {"amt_avg": 1500, "amt_std": 600, "fail_rate": [88, 8, 4]},24 "electronics": {"amt_avg": 18000, "amt_std": 12000, "fail_rate": [82, 12, 6]},25 "restaurant": {"amt_avg": 1000, "amt_std": 400, "fail_rate": [87, 9, 4]},26 "pharmacy": {"amt_avg": 800, "amt_std": 500, "fail_rate": [90, 7, 3]},27 "clothing": {"amt_avg": 3500, "amt_std": 2000, "fail_rate": [85, 10, 5]},28 "mobile": {"amt_avg": 3000, "amt_std": 4500, "fail_rate": [83, 11, 6]},29 "general_store": {"amt_avg": 400, "amt_std": 250, "fail_rate": [89, 8, 3]},30 "medical": {"amt_avg": 2500, "amt_std": 1500, "fail_rate": [91, 6, 3]},31}32 33HOURLY_WEIGHTS = {34 "default": [.3, .5, .7, .8, .9, .8, .7, .8, .9, 1.0, 1.2, 1.3, 1.1, .8, .4],35 "restaurant": [.2, .3, .4, .5, 1.3, 1.5, 1.0, .5, .4, .5, .7, 1.2, 1.5, 1.3, .8],36 "pharmacy": [.3, .5, .6, .7, .8, .9, .8, .7, .8, .9, 1.2, 1.4, 1.3, 1.0, .5],37 "supermarket": [.4, .6, .7, .8, .9, .8, .7, .7, .8, .9, 1.1, 1.4, 1.5, 1.2, .5],38 "electronics": [.3, .5, .7, .9, 1.0, 1.0, 1.0, 1.2, 1.3, 1.2, 1.0, .8, .6, .4, .2],39 "medical": [.5, .9, 1.3, 1.4, 1.0, .8, .7, .7, .6, .5, .5, .4, .3, .2, .1],40 "clothing": [.3, .5, .6, .7, .8, .9, 1.0, 1.2, 1.3, 1.2, 1.0, .8, .6, .4, .2],41 "mobile": [.3, .5, .7, .8, .8, .8, .9, 1.0, 1.2, 1.3, 1.2, 1.0, .8, .5, .3],42}43 44WALLET_TYPES = [45 ("jazzcash", 0.30), ("easypaisa", 0.25),46 ("sadapay", 0.10), ("nayapay", 0.08), ("raast", 0.27),47]48 49PAYMENT_MIX = {50 "supermarket": [("card", 0.50), ("bank_transfer", 0.25), ("wallet", 0.15), ("qr", 0.10)],51 "restaurant": [("card", 0.40), ("wallet", 0.25), ("qr", 0.25), ("bank_transfer", 0.10)],52 "electronics": [("card", 0.60), ("bank_transfer", 0.30), ("wallet", 0.10)],53 "pharmacy": [("card", 0.45), ("wallet", 0.30), ("bank_transfer", 0.20), ("qr", 0.05)],54 "clothing": [("card", 0.50), ("wallet", 0.20), ("bank_transfer", 0.20), ("qr", 0.10)],55 "mobile": [("wallet", 0.40), ("card", 0.30), ("bank_transfer", 0.20), ("qr", 0.10)],56 "general_store": [("wallet", 0.35), ("card", 0.25), ("qr", 0.25), ("bank_transfer", 0.15)],57 "medical": [("card", 0.55), ("bank_transfer", 0.30), ("wallet", 0.15)],58}59 60CARD_TYPES = [("card_visa", 0.55), ("card_mastercard", 0.45)]61FAILURE_REASONS = [62 ("bank_timeout", 0.30, "91"),63 ("insufficient_funds", 0.25, "51"),64 ("declined_by_issuer", 0.20, "05"),65 ("invalid_account", 0.10, "14"),66 ("expired_card", 0.08, "54"),67 ("limit_exceeded", 0.05, "61"),68 ("invalid_cvv", 0.02, "N7"),69]70 71RAAST_FAILURE_REASONS = [72 ("closed_account", 0.30, "AC04"),73 ("invalid_iban", 0.25, "AC01"),74 ("duplicate_transaction", 0.20, "URDF"),75 ("insufficient_funds", 0.15, "AM04"),76 ("limit_exceeded", 0.10, "AM06"),77]78 79 80def _pick_raast_failure_reason():81 """Raast/ISO20022-specific failure codes."""82 reason = _pick(RAAST_FAILURE_REASONS)83 code = next(f[2] for f in RAAST_FAILURE_REASONS if f[0] == reason)84 return reason, code85 86 87DOW_FACTOR = {0: 1.0, 1: 1.0, 2: 0.95, 3: 1.0, 4: 0.75, 5: 1.15, 6: 0.6}88 89_HOURLY_SUMS = {cat: sum(w) for cat, w in HOURLY_WEIGHTS.items()}90_DEFAULT_HOURLY_SUM = _HOURLY_SUMS["default"]91 92 93# ── Helpers ──────────────────────────────────────────────────────────────────94 95def _pick(pairs):96 return random.choices(97 [p[0] for p in pairs], weights=[p[1] for p in pairs], k=198 )[0]99 100 101def _pick_payment_method(category):102 group = _pick(PAYMENT_MIX.get(category, PAYMENT_MIX["general_store"]))103 if group == "card":104 return _pick(CARD_TYPES)105 if group == "wallet":106 return _pick(WALLET_TYPES)107 if group == "qr":108 return "qr_code"109 return "bank_transfer"110 111 112def _pick_failure_reason():113 reason = _pick(FAILURE_REASONS)114 code = next(f[2] for f in FAILURE_REASONS if f[0] == reason)115 return reason, code116 117 118def _pick_latency(is_failed=False):119 gw = int(max(50, random.gauss(200, 80)))120 bank = (121 int(max(100, random.gauss(8000, 3000)))122 if is_failed123 else int(max(200, random.gauss(1500, 500)))124 )125 return gw, bank126 127 128def _monthly_factor(date):129 if date.day <= 7:130 return 1.15131 if date.day >= 25:132 return 0.85133 return 1.0134 135 136# ── Event Data ───────────────────────────────────────────────────────────────137 138@dataclass139class SimEvent:140 id: str141 type: str142 params: dict143 affected_merchant_ids: list144 started_at: datetime145 duration_minutes: int146 ends_at: datetime147 status: str = "active"148 severity: str = "high"149 150 def to_dict(self):151 return {152 "id": self.id,153 "type": self.type,154 "params": self.params,155 "affected_count": len(self.affected_merchant_ids),156 "started_at": self.started_at.isoformat(),157 "duration_minutes": self.duration_minutes,158 "ends_at": self.ends_at.isoformat(),159 "status": self.status,160 "severity": self.severity,161 }162 163 164# ── Simulator Engine ─────────────────────────────────────────────────────────165 166class SimulatorEngine:167 def __init__(self):168 self.merchants = {}169 self.bank_index = {}170 self.clock: Optional[datetime] = None171 self.tick_count = 0172 self.total_txns = 0173 self.active_events: dict[str, SimEvent] = {}174 self.event_history: list[SimEvent] = []175 self.tick_interval = 30176 self.sim_minutes_per_tick = 3177 self._running = False178 self._thread: Optional[threading.Thread] = None179 self._cache_invalidator = None180 self._last_cleanup_tick = 0181 182 def _db_path(self):183 base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))184 return os.path.join(base_dir, DB_PATH)185 186 def load_merchants(self):187 from app.db import get_connection, release_connection188 conn = get_connection()189 try:190 cursor = conn.cursor()191 cursor.execute(192 "SELECT id, name, city, segment, category, status, acquiring_bank, "193 "integrator_type, base_daily_vol, base_avg_amount, mdr_rate "194 "FROM merchants"195 )196 rows = cursor.fetchall()197 cols = (198 [desc[0] for desc in cursor.description]199 if hasattr(cursor, "description")200 else []201 )202 for raw_row in rows:203 row = dict(zip(cols, raw_row)) if cols else dict(raw_row)204 mid = row["id"]205 cat = row["category"]206 cat_cfg = CATEGORY_CONFIG.get(cat, CATEGORY_CONFIG["general_store"])207 hourly_sum = _HOURLY_SUMS.get(cat, _DEFAULT_HOURLY_SUM)208 prob_per_min = float(row["base_daily_vol"]) / (hourly_sum * 60)209 self.merchants[mid] = {210 "id": mid,211 "name": row["name"],212 "city": row["city"],213 "segment": row["segment"],214 "category": cat,215 "status": row["status"],216 "acquiring_bank": row["acquiring_bank"],217 "integrator_type": row["integrator_type"],218 "base_daily_vol": float(row["base_daily_vol"]),219 "base_avg_amount": float(row["base_avg_amount"]),220 "amt_std": cat_cfg["amt_std"],221 "base_fail_rate": cat_cfg["fail_rate"],222 "prob_per_min": prob_per_min,223 "mdr_rate": float(row["mdr_rate"]),224 }225 bank = row["acquiring_bank"]226 if bank not in self.bank_index:227 self.bank_index[bank] = []228 self.bank_index[bank].append(mid)229 finally:230 release_connection(conn)231 print(232 f"Simulator: loaded {len(self.merchants)} merchants, "233 f"{len(self.bank_index)} banks"234 )235 236 # ── Startup ──────────────────────────────────────────────────────────────237 238 def start(self, cache_invalidator=None):239 if self._running:240 return241 self.load_merchants()242 self._cache_invalidator = cache_invalidator243 244 # Detect gap and backfill if needed.245 # When Space resumes after pause, DB has stale data.246 # Backfill fills missing days so dashboard looks correct.247 self.clock = self._detect_start_clock()248 249 self._running = True250 self._thread = threading.Thread(target=self._run_loop, daemon=True)251 self._thread.start()252 print(253 f"Simulator: started at {self.clock.isoformat()}, "254 f"tick every {self.tick_interval}s, "255 f"{self.sim_minutes_per_tick} sim-min/tick"256 )257 258 def stop(self):259 self._running = False260 if self._thread:261 self._thread.join(timeout=5)262 print("Simulator: stopped")263 264 def _detect_start_clock(self) -> datetime:265 """266 Find last transaction day in DB.267 If gap > 1 day vs today, backfill missing days.268 Prevents stale-data artifacts when Space resumes after pause.269 Handles any gap dynamically — no hardcoded dates.270 """271 from app.db import get_connection, release_connection272 273 now = datetime.now().replace(second=0, microsecond=0)274 275 try:276 conn = get_connection()277 try:278 cursor = conn.cursor()279 cursor.execute(280 "SELECT MAX(day) as last_day FROM transactions"281 )282 row = cursor.fetchone()283 last_day_str = row[0] if row else None284 finally:285 release_connection(conn)286 except Exception as e:287 print(f"Simulator: could not read last DB day — {e}")288 print("Simulator: starting from now without backfill")289 return now290 291 if not last_day_str:292 print("Simulator: no existing data — starting fresh from now")293 return now294 295 last_day = datetime.strptime(296 str(last_day_str)[:10], "%Y-%m-%d"297 ).replace(hour=23, minute=59, second=0)298 299 gap_days = (now.date() - last_day.date()).days300 301 print(302 f"Simulator: last DB day = {str(last_day_str)[:10]}, "303 f"today = {now.date()}, gap = {gap_days} day(s)"304 )305 306 if gap_days <= 1:307 print("Simulator: gap <= 1 day — starting normally")308 return now309 310 if gap_days > 60:311 print(312 f"Simulator: gap {gap_days}d too large to backfill — "313 f"starting from now, baselines normalize in ~20 sim-days"314 )315 return now316 317 # Gap is 2-60 days — backfill it318 print(f"Simulator: backfilling {gap_days} missing day(s)...")319 self._backfill_gap(last_day, now)320 print(f"Simulator: backfill complete — resuming from {now.date()}")321 return now322 323 def _backfill_gap(self, last_day: datetime, now: datetime) -> None:324 """325 Generate realistic transactions for each missing day between326 last_day and now. Uses same generation logic as live ticks.327 Fully dynamic — works for any gap size up to 60 days.328 No hardcoded dates anywhere.329 """330 current = (last_day + timedelta(days=1)).replace(331 hour=8, minute=0, second=0, microsecond=0332 )333 days_filled = 0334 335 while current.date() < now.date():336 day_str = current.strftime("%Y-%m-%d")337 all_txns = []338 339 # Generate one tick per hour for business hours 8am-10pm340 for hour in range(8, 23):341 self.clock = current.replace(hour=hour, minute=0)342 txns = self._generate_tick_transactions()343 all_txns.extend(txns)344 345 if all_txns:346 self._insert_transactions(all_txns)347 348 # Age settlements for this day before moving to next349 self.clock = current.replace(hour=23, minute=59)350 self._age_settlements()351 352 print(f" Backfilled {day_str}: {len(all_txns)} transactions")353 354 current += timedelta(days=1)355 days_filled += 1356 357 print(f"Simulator: {days_filled} day(s) backfilled successfully")358 359 # ── Main Loop ────────────────────────────────────────────────────────────360 361 def _run_loop(self):362 while self._running:363 try:364 self.tick()365 except Exception as e:366 print(f"Simulator tick error: {e}")367 time.sleep(self.tick_interval)368 369 def tick(self):370 self.tick_count += 1371 self.clock += timedelta(minutes=self.sim_minutes_per_tick)372 373 self._process_events()374 375 if self.tick_count % 10 == 0:376 self._maybe_spawn_event()377 378 txns = self._generate_tick_transactions()379 380 if txns:381 self._insert_transactions(txns)382 self.total_txns += len(txns)383 384 if self.tick_count % 100 == 0:385 self._cleanup_old_data()386 self._age_settlements()387 if self._cache_invalidator:388 self._cache_invalidator()389 print(390 f"Simulator: tick {self.tick_count}, "391 f"clock={self.clock.strftime('%H:%M')}, "392 f"+{len(txns)} txns, "393 f"{self.total_txns} total, "394 f"{len(self.active_events)} events"395 )396 397 # ── Event Management ─────────────────────────────────────────────────────398 399 def _process_events(self):400 expired = []401 for eid, event in self.active_events.items():402 if self.clock >= event.ends_at:403 event.status = "expired"404 expired.append(eid)405 self.event_history.append(event)406 print(f"Simulator: event expired: {event.type} ({eid})")407 for eid in expired:408 del self.active_events[eid]409 if expired and self._cache_invalidator:410 self._cache_invalidator()411 412 def _maybe_spawn_event(self):413 # Maintain minimum floor of active decline events414 active_declines = sum(415 1 for e in self.active_events.values()416 if e.type == "merchant_decline"417 )418 419 spawn_chance = 0.15420 if active_declines < 5:421 spawn_chance = 0.40422 423 if random.random() > spawn_chance:424 return425 426 if active_declines < 5:427 event_type = random.choices(428 ["merchant_decline", "bank_outage", "amount_anomaly",429 "merchant_recovery", "raast_adoption_surge"],430 weights=[0.50, 0.20, 0.10, 0.10, 0.10],431 k=1,432 )[0]433 else:434 event_type = random.choices(435 ["bank_outage", "merchant_decline", "amount_anomaly",436 "merchant_recovery", "raast_adoption_surge"],437 weights=[0.25, 0.25, 0.10, 0.20, 0.20],438 k=1,439 )[0]440 441 event = self._create_event(event_type)442 if event:443 self.active_events[event.id] = event444 if self._cache_invalidator:445 self._cache_invalidator()446 print(447 f"Simulator: auto event: {event.type} ({event.id}) "448 f"→ {len(event.affected_merchant_ids)} merchants"449 )450 451 def _create_event(452 self, event_type: str, params: dict = None453 ) -> Optional[SimEvent]:454 if event_type == "bank_outage":455 active_banks = [456 e.params.get("bank")457 for e in self.active_events.values()458 if e.type == "bank_outage"459 ]460 available = [b for b in BANKS if b not in active_banks]461 if not available:462 return None463 bank = params.get("bank") if params else random.choice(available)464 if bank not in self.bank_index:465 return None466 p = params or {}467 duration = p.get("duration_minutes", random.randint(30, 90))468 return SimEvent(469 id=str(uuid.uuid4())[:8],470 type="bank_outage",471 params={472 "bank": bank,473 "failure_rate_override": p.get(474 "failure_rate", round(random.uniform(0.30, 0.50), 2)475 ),476 "latency_multiplier": p.get(477 "latency_multiplier", round(random.uniform(4, 8), 1)478 ),479 },480 affected_merchant_ids=list(self.bank_index.get(bank, [])),481 started_at=self.clock,482 duration_minutes=duration,483 ends_at=self.clock + timedelta(minutes=duration),484 severity="high",485 )486 487 elif event_type == "merchant_decline":488 declining_ids = set()489 for e in self.active_events.values():490 if e.type in ("merchant_decline", "merchant_recovery"):491 declining_ids.update(e.affected_merchant_ids)492 available = [493 mid for mid in self.merchants494 if mid not in declining_ids495 and self.merchants[mid]["status"] == "active"496 ]497 if not available:498 return None499 p = params or {}500 mid = p.get("merchant_id") if params else random.choice(available)501 if mid not in self.merchants:502 return None503 duration = p.get(504 "duration_minutes", random.randint(3 * 1440, 7 * 1440)505 )506 return SimEvent(507 id=str(uuid.uuid4())[:8],508 type="merchant_decline",509 params={510 "merchant_id": mid,511 "decline_rate": p.get(512 "decline_rate", round(random.uniform(0.08, 0.15), 2)513 ),514 },515 affected_merchant_ids=[mid],516 started_at=self.clock,517 duration_minutes=duration,518 ends_at=self.clock + timedelta(minutes=duration),519 severity="medium",520 )521 522 elif event_type == "amount_anomaly":523 p = params or {}524 mid = p.get(525 "merchant_id",526 random.choice(527 [m for m in self.merchants528 if self.merchants[m]["status"] == "active"]529 ),530 )531 if mid not in self.merchants:532 return None533 duration = p.get("duration_minutes", random.randint(30, 120))534 return SimEvent(535 id=str(uuid.uuid4())[:8],536 type="amount_anomaly",537 params={538 "merchant_id": mid,539 "amount_multiplier": p.get(540 "amount_multiplier", round(random.uniform(8, 15), 1)541 ),542 },543 affected_merchant_ids=[mid],544 started_at=self.clock,545 duration_minutes=duration,546 ends_at=self.clock + timedelta(minutes=duration),547 severity="high",548 )549 550 elif event_type == "merchant_recovery":551 declining = [552 e for e in self.active_events.values()553 if e.type == "merchant_decline"554 ]555 if not declining:556 return None557 source = random.choice(declining)558 mid = source.params["merchant_id"]559 source.status = "expired"560 source.ends_at = self.clock561 self.event_history.append(source)562 del self.active_events[source.id]563 p = params or {}564 duration = p.get(565 "duration_minutes", random.randint(2 * 1440, 5 * 1440)566 )567 return SimEvent(568 id=str(uuid.uuid4())[:8],569 type="merchant_recovery",570 params={571 "merchant_id": mid,572 "recovery_rate": p.get(573 "recovery_rate", round(random.uniform(0.05, 0.10), 2)574 ),575 },576 affected_merchant_ids=[mid],577 started_at=self.clock,578 duration_minutes=duration,579 ends_at=self.clock + timedelta(minutes=duration),580 severity="low",581 )582 583 elif event_type == "raast_adoption_surge":584 active_ids = set()585 for e in self.active_events.values():586 if e.type == "raast_adoption_surge":587 active_ids.update(e.affected_merchant_ids)588 available = [589 mid for mid in self.merchants590 if mid not in active_ids591 and self.merchants[mid]["status"] == "active"592 ]593 if not available:594 return None595 p = params or {}596 count = p.get("merchant_count", random.randint(30, 80))597 affected = random.sample(available, min(count, len(available)))598 duration = p.get(599 "duration_minutes", random.randint(3 * 1440, 7 * 1440)600 )601 return SimEvent(602 id=str(uuid.uuid4())[:8],603 type="raast_adoption_surge",604 params={605 "raast_mult": p.get(606 "raast_mult", round(random.uniform(3, 6), 1)607 ),608 "merchant_count": len(affected),609 },610 affected_merchant_ids=affected,611 started_at=self.clock,612 duration_minutes=duration,613 ends_at=self.clock + timedelta(minutes=duration),614 severity="low",615 )616 617 return None618 619 def _get_event_modifiers(self, merchant_id: int):620 """Returns (vol_mult, fail_rate_override, amt_mult, latency_mult, raast_boost)."""621 vol_mult = 1.0622 fail_override = None623 amt_mult = 1.0624 lat_mult = 1.0625 raast_boost = False626 627 for event in self.active_events.values():628 if merchant_id not in event.affected_merchant_ids:629 continue630 if event.type == "bank_outage":631 fail_override = event.params["failure_rate_override"]632 lat_mult = max(lat_mult, event.params["latency_multiplier"])633 vol_mult *= 0.7634 elif event.type == "merchant_decline":635 mins_elapsed = (636 self.clock - event.started_at637 ).total_seconds() / 60638 days_elapsed = mins_elapsed / 1440639 decline = event.params["decline_rate"] * days_elapsed640 vol_mult *= max(0.05, 1.0 - decline)641 elif event.type == "amount_anomaly":642 if event.params["merchant_id"] == merchant_id:643 amt_mult = event.params["amount_multiplier"]644 elif event.type == "merchant_recovery":645 if event.params["merchant_id"] == merchant_id:646 mins_elapsed = (647 self.clock - event.started_at648 ).total_seconds() / 60649 days_elapsed = mins_elapsed / 1440650 recovery = event.params["recovery_rate"] * days_elapsed651 vol_mult *= min(1.0, 0.3 + recovery)652 elif event.type == "raast_adoption_surge":653 raast_boost = True654 655 return vol_mult, fail_override, amt_mult, lat_mult, raast_boost656 657 # ── Transaction Generation ────────────────────────────────────────────────658 659 def _generate_tick_transactions(self) -> list:660 hour = self.clock.hour661 if hour < 8 or hour >= 23:662 return []663 664 hour_idx = hour - 8665 dow = self.clock.weekday()666 day_str = self.clock.strftime("%Y-%m-%d")667 dow_mult = DOW_FACTOR.get(dow, 1.0)668 monthly_mult = _monthly_factor(self.clock)669 time_mult = dow_mult * monthly_mult670 671 txns = []672 673 for mid, m in self.merchants.items():674 if m["status"] != "active":675 continue676 677 weights = HOURLY_WEIGHTS.get(m["category"], HOURLY_WEIGHTS["default"])678 hourly_w = weights[hour_idx]679 vol_mult, fail_override, amt_mult, lat_mult, raast_boost = (680 self._get_event_modifiers(mid)681 )682 683 expected = m["prob_per_min"] * hourly_w * time_mult * vol_mult684 count = int(expected) + (685 1 if random.random() < (expected - int(expected)) else 0686 )687 if count <= 0:688 continue689 690 for _ in range(count):691 if fail_override is not None:692 status = random.choices(693 ["success", "failed", "declined"],694 weights=[695 1 - fail_override,696 fail_override * 0.7,697 fail_override * 0.3,698 ],699 k=1,700 )[0]701 else:702 status = random.choices(703 ["success", "failed", "declined"],704 weights=m["base_fail_rate"],705 k=1,706 )[0]707 708 amount = round(709 max(50, random.gauss(m["base_avg_amount"] * amt_mult, m["amt_std"])),710 2,711 )712 ts = self.clock.replace(second=random.randint(0, 59)).isoformat()713 payment_method = _pick_payment_method(m["category"])714 715 if raast_boost and payment_method != "raast" and random.random() < 0.7:716 payment_method = "raast"717 718 if status == "success":719 failure_reason = None720 response_code = "00"721 settlement = "settled" if payment_method == "raast" else "pending"722 else:723 if payment_method == "raast":724 failure_reason, response_code = _pick_raast_failure_reason()725 else:726 failure_reason, response_code = _pick_failure_reason()727 settlement = "pending"728 729 gw_lat, bank_lat = _pick_latency(status != "success")730 bank_lat = int(bank_lat * lat_mult)731 732 txns.append((733 mid, amount, status, ts, day_str,734 payment_method, failure_reason, response_code,735 gw_lat, bank_lat, settlement,736 ))737 738 return txns739 740 def _insert_transactions(self, txns: list):741 from app.db import get_connection, release_connection742 conn = get_connection()743 try:744 cursor = conn.cursor()745 is_pg = hasattr(conn, "autocommit") and os.getenv("DATABASE_URL")746 if is_pg:747 from psycopg2.extras import execute_values748 execute_values(749 cursor,750 """751 INSERT INTO transactions (752 merchant_id, amount, status, timestamp, day,753 payment_method, failure_reason, response_code,754 gateway_latency_ms, bank_latency_ms, settlement_status755 ) VALUES %s756 """,757 txns,758 )759 else:760 cursor.executemany(761 "INSERT INTO transactions ("762 "merchant_id, amount, status, timestamp, day, "763 "payment_method, failure_reason, response_code, "764 "gateway_latency_ms, bank_latency_ms, settlement_status"765 ") VALUES (?,?,?,?,?,?,?,?,?,?,?)",766 txns,767 )768 conn.commit()769 finally:770 release_connection(conn)771 772 # ── Maintenance ───────────────────────────────────────────────────────────773 774 def _cleanup_old_data(self):775 # DISABLED for demo deployment.776 return777 778 def _age_settlements(self):779 from app.db import get_connection, release_connection780 cutoff = (self.clock - timedelta(days=2)).strftime("%Y-%m-%d")781 782 # Deterministic held merchant set — bottom 3% of IDs783 # Simulates compliance holds / bank processing delays784 all_ids = sorted(self.merchants.keys())785 held_count = max(3, len(all_ids) // 33)786 held_merchant_ids = set(all_ids[:held_count])787 788 conn = get_connection()789 try:790 cursor = conn.cursor()791 is_pg = hasattr(conn, "autocommit") and os.getenv("DATABASE_URL")792 p = "%s" if is_pg else "?"793 794 if held_merchant_ids and is_pg:795 placeholders = ",".join([p] * len(held_merchant_ids))796 cursor.execute(797 f"""798 UPDATE transactions799 SET settlement_status = 'settled'800 WHERE day <= {p}801 AND status = 'success'802 AND settlement_status = 'pending'803 AND merchant_id NOT IN ({placeholders})804 """,805 (cutoff, *held_merchant_ids),806 )807 else:808 cursor.execute(809 f"""810 UPDATE transactions811 SET settlement_status = 'settled'812 WHERE day <= {p}813 AND status = 'success'814 AND settlement_status = 'pending'815 """,816 (cutoff,),817 )818 819 settled = cursor.rowcount820 821 cursor.execute(822 f"""823 UPDATE transactions824 SET settlement_status = 'reversed'825 WHERE day <= {p}826 AND status != 'success'827 AND settlement_status = 'pending'828 """,829 (cutoff,),830 )831 reversed_count = cursor.rowcount832 conn.commit()833 834 if settled or reversed_count:835 print(836 f"Simulator: settlements aged — {settled} settled, "837 f"{reversed_count} reversed, "838 f"{len(held_merchant_ids)} merchants held"839 )840 finally:841 release_connection(conn)842 843 # ── Manual Event Control ──────────────────────────────────────────────────844 845 def trigger_event(self, event_type: str, params: dict) -> dict:846 event = self._create_event(event_type, params)847 if not event:848 return {849 "error": f"Cannot create {event_type} event — "850 "bank not found, merchant not found, or conflicting event active"851 }852 self.active_events[event.id] = event853 if self._cache_invalidator:854 self._cache_invalidator()855 print(f"Simulator: manual event: {event.type} ({event.id})")856 return event.to_dict()857 858 def resolve_event(self, event_id: str) -> dict:859 event = self.active_events.get(event_id)860 if not event:861 return {"error": f"Event not found: {event_id}"}862 event.status = "resolved"863 event.ends_at = self.clock864 self.event_history.append(event)865 del self.active_events[event_id]866 print(f"Simulator: resolved: {event.type} ({event_id})")867 return event.to_dict()868 869 def set_speed(870 self, tick_interval: int = None, sim_minutes_per_tick: int = None871 ):872 if tick_interval is not None:873 self.tick_interval = max(1, min(60, tick_interval))874 if sim_minutes_per_tick is not None:875 self.sim_minutes_per_tick = max(1, min(1440, sim_minutes_per_tick))876 return {877 "tick_interval_seconds": self.tick_interval,878 "sim_minutes_per_tick": self.sim_minutes_per_tick,879 }880 881 # ── Status ────────────────────────────────────────────────────────────────882 883 def get_status(self) -> dict:884 return {885 "running": self._running,886 "clock": self.clock.isoformat() if self.clock else None,887 "tick_count": self.tick_count,888 "total_txns_generated": self.total_txns,889 "tick_interval_seconds": self.tick_interval,890 "sim_minutes_per_tick": self.sim_minutes_per_tick,891 "active_events": len(self.active_events),892 "history_events": len(self.event_history),893 "merchants_loaded": len(self.merchants),894 }895 896 def get_events(self) -> dict:897 return {898 "active": [e.to_dict() for e in self.active_events.values()],899 "recent": [900 e.to_dict() for e in reversed(self.event_history[-20:])901 ],902 }903 904 905# ── Global Instance ───────────────────────────────────────────────────────────906 907sim = SimulatorEngine()