pg_trickle — Real-time Financial Risk Pipeline

""" pg_trickle demo — financial risk pipeline scenario: HTML, DAG, and data queries. Exposes: HTML — complete page HTML with {{ dag }} placeholder DAG_DIAGRAM — ASCII art DAG to substitute into {{ dag }} get_data(conn, safe_query, serialize) → dict for /api/data """ # ── Full page HTML ───────────────────────────────────────────────────────────── HTML = r""" pg_trickle — Real-time Financial Risk Pipeline

pg_trickle Real-time Financial Risk Pipeline

connecting…

—

Active Positions

—

Total Book Exposure

—

Unrealized P&L

—

Total Book VaR (95%)

🚨 Capital Breach Dashboard L10 DIFFERENTIAL showcase — LIMIT 10, change ratio ~0.02–0.1

#	Portfolio	Manager	Strategy	Book Value	VaR 95%	Req. Capital	Capital %	Status

10-level DAG: trades → net_positions → position_values → account_pnl → portfolio_pnl → sector_exposure → var_contributions → account_var → portfolio_var → regulatory_capital → breach_dashboard. Fixed cardinality (10 rows): rank shifts are rare — KEEP DIFFERENTIAL.

🏭 Sector Exposure L5

Sector	Instruments	Exposure	P&L	% Book

💼 Portfolio P&L L4

Portfolio	Manager	Strategy	Accounts	Book Value	Unrealized P&L	Return %

📊 Portfolio VaR L8

Portfolio	Strategy	Book Value	VaR 95%	Stressed VaR 99%	VaR % NAV

📡 Live Price Snapshot L1 DIFFERENTIAL showcase — 1 of 30 instruments ticks per cycle, change ratio ~0.033

Ticker	Name	Sector	Bid	Ask	Mid	Base	Drift $	Drift %	Updated

Generator ticks one instrument per cycle (~1 s). Only that instrument's positions cascade through all 10 DAG levels. Change ratio at L2 ≈ 0.02. Mode Advisor recommends KEEP DIFFERENTIAL at every level.

⚡ Stream Table Status

Name	Level	Mode	Schedule	Status

📊 DAG Topology (10 levels) — click to expand

{{ dag }}

🏥 Stream Table Health — pgtrickle.pgt_status()

Name	Mode	Schedule	Populated	Last Refresh	Duration	Rows Δ	Status

🌳 Dependency Tree — pgtrickle.dependency_tree()

loading…

📜 Refresh History — pgtrickle.pgt_refresh_history

Table	Action	Started	ms	Rows Δ	Status

⚡ Refresh Efficiency — pgtrickle.refresh_efficiency()

Stream Table	Total Refreshes	DIFF	FULL	Avg DIFF ms	Avg FULL ms	DIFF Speedup	Avg Δ Ratio

🎯 Refresh Mode Advisor — pgtrickle.refresh_mode_advisor()

Stream Table	Current	Effective	Recommendation	Confidence	Reason

""" # ── DAG diagram ──────────────────────────────────────────────────────────────── DAG_DIAGRAM = r""" Base tables (LEAF schedule) L1 ── CALCULATED downstream ────────────────────────────────────────────────────────────── ┌─────────────────┐ 2s ┌──────────────────┐ │ market_prices │──────────►│ price_snapshot │─────────────────────────────────────────────────────────────────────┐ │ (tick: 1/30 │ │ (DIFF, 2s) │ │ │ instruments │ └──────────────────┘ │ │ per cycle) │ │ └─────────────────┘ │ │ L2 ┌─────────────────┐ 1s ┌──────────────────┐ ┌──────────────────┐ ▼ │ trades │──────────►│ net_positions │───────────►│ position_values │ (DIFF, calc) joins price_snapshot │ (append-only │ │ (DIFF, 1s) │ │ MV = qty × price │ │ order stream) │ └──────────────────┘ └────────┬─────────┘ └─────────────────┘ │ │ L3 L5 ├──────────────────────────►┌──────────────────┐ │ │ sector_exposure │ (DIFF, calc) ▼ └──────────────────┘ ┌──────────────────┐ │ account_pnl │ (DIFF, calc) └────────┬─────────┘ │ L4 ▼ L6 ┌──────────────────┐ position_values │ portfolio_pnl │ ──────────────►┌──────────────────┐ │ (DIFF, calc) │ │ var_contributions│ (DIFF, calc) └──────────────────┘ └────────┬─────────┘ │ L7 ▼ ┌──────────────────┐ │ account_var │ (DIFF, calc) └────────┬─────────┘ │ L8 ▼ ┌──────────────────┐ │ portfolio_var │ (DIFF, calc) └────────┬─────────┘ │ L9 ▼ ┌──────────────────────┐ │ regulatory_capital │ (DIFF, calc) └────────┬─────────────┘ │ L10 ▼ ┌──────────────────────┐ │ breach_dashboard │ ← DIFF SHOWCASE │ LIMIT 10 portfolios │ change ratio ~0.02 └──────────────────────┘ Only market_prices (2s) and trades (1s) have CALCULATED schedules. All 9 derived levels propagate via schedule => 'calculated'. One price tick → ~30 of 1,500 positions change (ratio ≈ 0.02). """ # ── Data queries ─────────────────────────────────────────────────────────────── def get_data(conn, safe_query, serialize) -> dict: price_snapshot = safe_query(conn, """ SELECT ticker, instrument_name, sector, asset_class, bid, ask, mid_price, base_price, price_drift, drift_pct, price_updated_at FROM price_snapshot ORDER BY ticker """) position_values = safe_query(conn, """ SELECT account_id, account_name, portfolio_id, ticker, sector, asset_class, net_quantity, avg_cost_basis, mid_price, market_value, unrealized_pnl, pnl_pct FROM position_values WHERE market_value IS NOT NULL ORDER BY ABS(market_value) DESC NULLS LAST """) sector_exposure = safe_query(conn, """ SELECT sector, asset_class, instrument_count, account_count, total_exposure, total_unrealized_pnl, sector_return_pct, pct_of_total_book FROM sector_exposure ORDER BY ABS(total_exposure) DESC NULLS LAST """) portfolio_pnl = safe_query(conn, """ SELECT portfolio_id, portfolio_name, manager, strategy, account_count, total_positions, total_market_value, total_unrealized_pnl, portfolio_return_pct FROM portfolio_pnl ORDER BY total_market_value DESC NULLS LAST """) portfolio_var = safe_query(conn, """ SELECT portfolio_id, portfolio_name, manager, strategy, total_market_value, total_pnl, sum_var_95, sum_stressed_var_99, portfolio_var_95, var_pct_nav FROM portfolio_var ORDER BY portfolio_var_95 DESC NULLS LAST """) breach_dashboard = safe_query(conn, """ SELECT rank, portfolio_name, manager, strategy, total_market_value, total_pnl, portfolio_var_95, required_capital, capital_limit, capital_headroom, capital_ratio_pct, capital_status FROM breach_dashboard ORDER BY rank """) st_status = safe_query(conn, """ SELECT name, status, refresh_mode, is_populated, schedule FROM pgtrickle.pgt_status() ORDER BY name """) return { "price_snapshot": serialize(price_snapshot), "position_values": serialize(position_values), "sector_exposure": serialize(sector_exposure), "portfolio_pnl": serialize(portfolio_pnl), "portfolio_var": serialize(portfolio_var), "breach_dashboard": serialize(breach_dashboard), "st_status": serialize(st_status), }