"""InFinBot: read-only rankings of recorded dispositions by recorded worker and asset class. No broker verification, parameter changes, notifications or capital clearance. """ from pathlib import Path import argparse import json import sqlite3 from datetime import datetime, timezone from decimal import Decimal, InvalidOperation, localcontext def amount(value): if isinstance(value, bool) or value is None: raise ValueError("Missing or invalid monetary amount") try: result = Decimal(str(value)) except InvalidOperation: raise ValueError("Invalid monetary amount") from None if not result.is_finite() or abs(result) > Decimal("1e20"): raise ValueError("Nonfinite or out-of-range monetary amount") if result.as_tuple().exponent < -12: raise ValueError("Monetary precision exceeds 12 decimal places") return result def when(value): if not isinstance(value, str) or len(value) > 40: raise ValueError("Missing or invalid disposition date") result = datetime.fromisoformat(value.replace("Z", "+00:00")) return result.replace(tzinfo=timezone.utc) if result.tzinfo is None else result.astimezone(timezone.utc) def summarize(values): count = len(values) wins, losses, flat = sum(x > 0 for x in values), sum(x < 0 for x in values), sum(x == 0 for x in values) profits = sum((x for x in values if x > 0), Decimal(0)) gross_losses = -sum((x for x in values if x < 0), Decimal(0)) return {"total_trades":count, "wins":wins, "losses":losses, "breakeven":flat, "win_rate_pct":str(Decimal(wins)*100/count) if count else None, "gross_profits":str(profits), "gross_losses":str(gross_losses), "net_pnl":str(profits-gross_losses), "interpretation":"RECORDED_SAMPLE_DIAGNOSTIC_ONLY"} class MultiWorkerPerformanceRanker: def __init__(self, db_filepath): self.db_filepath = str(Path(db_filepath).resolve()) self.conn = None if not Path(self.db_filepath).is_file(): raise FileNotFoundError("Existing ledger required; no database created") self.conn = sqlite3.connect(Path(self.db_filepath).as_uri()+"?mode=ro", uri=True, timeout=1) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA query_only=ON") def execute_worker_ranking_audit(self, execution_tag=None): report = {"status":"EVIDENCE_UNAVAILABLE", "reported_at":datetime.now(timezone.utc).isoformat(), "database":self.db_filepath, "execution_tag_scope":execution_tag, "scoreboard":[], "portfolio":None, "worker_attribution":"RECORDED_ORDER_JOIN_REQUIRED", "live_executions_verified":False, "parameters_updated":False, "ledger_modified":False, "emails_sent":0, "live_clearance_granted":False} try: if execution_tag is not None and (not isinstance(execution_tag, str) or not execution_tag.strip() or len(execution_tag)>150): raise ValueError("Execution tag must be a nonempty recorded scope of at most 150 characters") self.conn.execute("BEGIN") if [r[0] for r in self.conn.execute("PRAGMA quick_check")] != ["ok"]: raise ValueError("SQLite structural check failed") required = {"form_8949_ledger":{"order_id","equity_symbol","date_acquired","date_sold_or_disposed", "gross_proceeds","cost_basis","net_gain_or_loss"}, "simulated_orders":{"order_id","execution_tag","asset_class","equity_symbol"}} for table, columns in required.items(): present = {r[1] for r in self.conn.execute("PRAGMA table_info("+table+")")} if not columns <= present: raise ValueError("Missing compatible "+table+"; attribution cannot be guessed from symbols") if self.conn.execute("SELECT order_id FROM simulated_orders GROUP BY order_id HAVING COUNT(*)<>1 OR order_id IS NULL LIMIT 1").fetchone(): raise ValueError("Ambiguous order identifiers") if self.conn.execute("SELECT order_id FROM form_8949_ledger GROUP BY order_id HAVING COUNT(*)<>1 OR order_id IS NULL LIMIT 1").fetchone(): raise ValueError("Ambiguous disposition identifiers; partial-lot accounting adapter required") rows = self.conn.execute('''SELECT l.*, s.execution_tag AS recorded_tag, s.asset_class AS recorded_asset, s.equity_symbol AS order_symbol FROM form_8949_ledger l LEFT JOIN simulated_orders s ON s.order_id=l.order_id ORDER BY l.order_id''').fetchall() groups, totals = {}, [] now = datetime.now(timezone.utc) # Validate all rows before filtering. Hidden orphan rows cannot silently vanish from a scoped audit. with localcontext() as context: context.prec = 60 for row in rows: tag, asset = row["recorded_tag"], row["recorded_asset"] if not isinstance(tag,str) or not tag.strip() or len(tag)>150: raise ValueError("Disposition has no valid recorded worker attribution") if asset not in ("CRYPTO","FUTURES","STOCKS"): raise ValueError("Disposition has no supported recorded asset class") if not row["equity_symbol"] or row["equity_symbol"] != row["order_symbol"]: raise ValueError("Disposition symbol disagrees with recorded order") acquired, sold = when(row["date_acquired"]), when(row["date_sold_or_disposed"]) if sold < acquired or sold > now: raise ValueError("Disposition dates reversed or in the future") proceeds, basis, pnl = map(amount,(row["gross_proceeds"],row["cost_basis"],row["net_gain_or_loss"])) if proceeds<0 or basis<0: raise ValueError("Negative proceeds or basis") if "wash_sale_disallowed_amount" in row.keys() and amount(row["wash_sale_disallowed_amount"]) != 0: raise ValueError("Adjusted tax PnL requires a reviewed economic-return mapping") if "signed_adjustment_amount" in row.keys() and amount(row["signed_adjustment_amount"]) != 0: raise ValueError("Adjusted tax PnL requires a reviewed economic-return mapping") if pnl != proceeds-basis: raise ValueError("Economic PnL does not equal proceeds minus basis") if execution_tag is not None and tag != execution_tag: continue groups.setdefault((tag,asset),[]).append(pnl) totals.append(pnl) stats = [{"worker_name":tag, "execution_tag":tag, "asset_class":asset, **summarize(values)} for (tag,asset),values in groups.items()] stats.sort(key=lambda x:(-Decimal(x["win_rate_pct"]),x["execution_tag"],x["asset_class"])) for rank, stat in enumerate(stats,1): stat["rank"] = rank report.update(status="RECORDED_RANKING_COMPLETE" if totals else "NO_DISPOSITION_EVIDENCE", scoreboard=stats, portfolio=summarize(totals) if totals else None, worker_attribution="RECORDED_ORDER_JOIN_NOT_BROKER_VERIFIED", date_semantics="NAIVE_DATES_INTERPRETED_AS_UTC; FUTURE_DISPOSITIONS_REJECTED", portfolio_net_pnl=str(sum(totals,Decimal(0))) if totals else None, portfolio_win_rate=str(Decimal(sum(x>0 for x in totals))*100/len(totals)) if totals else None) except (ValueError, sqlite3.Error, InvalidOperation) as exc: report["reason"] = str(exc) finally: self.conn.rollback() return report def close(self): if self.conn: self.conn.close() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--db-path", required=True, help="Existing current ledger; no default database or mock data") parser.add_argument("--execution-tag", help="Exact recorded worker scope; omitted means all recorded workers") args = parser.parse_args() ranker = None try: ranker = MultiWorkerPerformanceRanker(args.db_path) report = ranker.execute_worker_ranking_audit(args.execution_tag) except (OSError, sqlite3.Error, ValueError) as exc: report = {"status":"EVIDENCE_UNAVAILABLE", "reason":str(exc), "scoreboard":[], "portfolio":None, "ledger_modified":False, "parameters_updated":False, "live_clearance_granted":False} finally: if ranker: ranker.close() report.update(engine_architecture="INFINBOT_WORKER_PERFORMANCE_RANKER", data_origin="RECORDED_LEDGER_NOT_BROKER_VERIFIED", trial_days_credited=0) print(json.dumps(report, indent=2, allow_nan=False)) return 0 if report["status"] == "RECORDED_RANKING_COMPLETE" else 1 if __name__ == "__main__": raise SystemExit(main())