From 14a9b4fcc1c02b7976bf64a8bb9f80730963fe56 Mon Sep 17 00:00:00 2001 From: Celes Renata Date: Fri, 3 Jul 2026 08:29:24 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20ops=20pipeline=20=E2=80=94=20rate=20limi?= =?UTF-8?q?t,=20validation=20cycle,=20snapshots,=20config=20query,=20rejec?= =?UTF-8?q?tion=20reason?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Polygon rate limit env-configurable, default 5 (free tier) - v3_engine_enabled reads from JSONB config column correctly - Validation cycle (outcome eval + metrics) wired into scheduler hourly - Daily portfolio/risk snapshots after 16:30 ET with idempotency - Prediction snapshot price fallback from positions table - Order rejection_reason + rejected_at persisted on insert - Lake-publisher scaled to 0 replicas (redundant) - Test fix: isolate per-type rate limit tests from Polygon global --- infra/helm/stonks-oracle/values.yaml | 2 +- services/adapters/broker_service.py | 10 +- services/aggregation/worker.py | 11 +- services/scheduler/app.py | 111 ++++++++++++++++++++- services/validation/prediction_snapshot.py | 14 ++- tests/test_scheduler_unit.py | 10 +- 6 files changed, 145 insertions(+), 13 deletions(-) diff --git a/infra/helm/stonks-oracle/values.yaml b/infra/helm/stonks-oracle/values.yaml index 7583dc4..ca8150b 100644 --- a/infra/helm/stonks-oracle/values.yaml +++ b/infra/helm/stonks-oracle/values.yaml @@ -139,7 +139,7 @@ services: limits: { cpu: 500m, memory: 256Mi } lakePublisher: - replicas: 1 + replicas: 0 pipeline: true image: lake-publisher command: "python -m services.lake_publisher.jobs" diff --git a/services/adapters/broker_service.py b/services/adapters/broker_service.py index 6421dd7..da883a2 100644 --- a/services/adapters/broker_service.py +++ b/services/adapters/broker_service.py @@ -105,12 +105,12 @@ INSERT INTO orders ( id, recommendation_id, broker_account_id, ticker, side, order_type, quantity, limit_price, stop_price, status, idempotency_key, broker_order_id, decision_trace, submitted_at, filled_at, - fill_price, fill_quantity + fill_price, fill_quantity, rejection_reason, rejected_at ) VALUES ( $1::uuid, $2, $3::uuid, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13::jsonb, $14, $15, - $16, $17 + $16, $17, $18, $19 ) ON CONFLICT (idempotency_key) DO UPDATE SET status = EXCLUDED.status, @@ -118,6 +118,8 @@ ON CONFLICT (idempotency_key) DO UPDATE SET filled_at = EXCLUDED.filled_at, fill_price = EXCLUDED.fill_price, fill_quantity = EXCLUDED.fill_quantity, + rejection_reason = COALESCE(EXCLUDED.rejection_reason, orders.rejection_reason), + rejected_at = COALESCE(EXCLUDED.rejected_at, orders.rejected_at), updated_at = NOW() """ @@ -361,6 +363,8 @@ async def persist_order( """Persist order, events, and risk evaluation to PostgreSQL.""" now = datetime.now(timezone.utc) filled_at = now if resp.status == OrderStatus.FILLED else None + rejection_reason = resp.error if resp.status == OrderStatus.REJECTED else None + rejected_at = now if resp.status == OrderStatus.REJECTED else None decision_trace = { "risk_evaluation": risk_eval, @@ -389,6 +393,8 @@ async def persist_order( filled_at, resp.filled_avg_price, resp.filled_quantity, + rejection_reason, + rejected_at, ) # Record order events diff --git a/services/aggregation/worker.py b/services/aggregation/worker.py index 64baeb1..49d2e46 100644 --- a/services/aggregation/worker.py +++ b/services/aggregation/worker.py @@ -316,19 +316,22 @@ async def fetch_probabilistic_scoring_enabled(pool: asyncpg.Pool) -> bool: # --------------------------------------------------------------------------- _V3_ENGINE_FLAG_QUERY = """ -SELECT value FROM risk_configs WHERE key = 'v3_engine_enabled' +SELECT config->>'v3_engine_enabled' AS enabled +FROM risk_configs +WHERE name = 'default' AND active = TRUE +LIMIT 1 """ async def _read_v3_flag(pool: asyncpg.Pool) -> bool: - """Read v3_engine_enabled from risk_configs. Default False on error. + """Read v3_engine_enabled from risk_configs JSONB. Default False on error. Requirements: 19.3, 19.6 """ try: row = await pool.fetchrow(_V3_ENGINE_FLAG_QUERY) - if row: - return str(row["value"]).lower() in ("true", "1", "yes") + if row and row["enabled"] is not None: + return row["enabled"].lower() in ("true", "1", "yes") return False except Exception as e: logger.warning("Failed to read v3_engine_enabled flag: %s", e) diff --git a/services/scheduler/app.py b/services/scheduler/app.py index 969efe3..e689341 100644 --- a/services/scheduler/app.py +++ b/services/scheduler/app.py @@ -75,7 +75,7 @@ DEFAULT_RATE_LIMITS: dict[str, int] = { # market_api + news_api share a single Polygon API key, so we cap the combined # throughput to stay safely under the plan limit. POLYGON_SOURCE_TYPES: set[str] = {"market_api", "news_api"} -POLYGON_GLOBAL_RATE_LIMIT: int = 45 +POLYGON_GLOBAL_RATE_LIMIT: int = int(os.getenv("POLYGON_GLOBAL_RATE_LIMIT", "5")) # How long to wait before retrying a failed source (seconds) DEFAULT_BACKOFF_BASE: int = 60 @@ -89,6 +89,13 @@ SCHEDULER_TICK: int = 15 # 15s tick × 60 cycles = 15 minutes AGGREGATION_CYCLE_INTERVAL: int = 60 +# Periodic snapshot capture: every N cycles (15s tick × 240 = ~60 minutes) +SNAPSHOT_CYCLE_INTERVAL = int(os.getenv("SNAPSHOT_CYCLE_INTERVAL", "240")) + +# Periodic validation cycle: evaluate matured predictions + compute metric snapshots +# 15s tick × 240 cycles = ~60 minutes +VALIDATION_CYCLE_INTERVAL = int(os.getenv("VALIDATION_CYCLE_INTERVAL", "240")) + def get_cadence_for_source(source_type: str, config: Optional[dict[str, Any]]) -> int: """Return the polling interval for a source. @@ -682,6 +689,96 @@ async def enqueue_periodic_aggregation(pool: asyncpg.Pool, rds: aioredis.Redis) return count +async def run_validation_cycle(pool: asyncpg.Pool) -> None: + """Run outcome evaluation and metric computation (hourly). + + Requirements: 2.3, 2.4, 2.5 + """ + from services.validation.metrics import compute_and_store_metric_snapshots + from services.validation.outcome_evaluator import evaluate_matured_predictions + + try: + outcomes = await evaluate_matured_predictions(pool) + logger.info("Validation: evaluated %d prediction outcomes", outcomes) + except Exception: + logger.exception("Validation: outcome evaluation failed") + return # Skip metrics if outcomes failed + + try: + snapshots = await compute_and_store_metric_snapshots(pool) + logger.info("Validation: computed %d metric snapshots", len(snapshots)) + except Exception: + logger.exception("Validation: metric computation failed") + + +async def maybe_capture_daily_snapshots(pool: asyncpg.Pool) -> None: + """Capture portfolio and risk snapshots once daily after market close. + + Requirements: 2.6, 2.7 + """ + et_now = datetime.now(ZoneInfo("America/New_York")) + + # Only after 4:30 PM ET + if et_now.hour < 16 or (et_now.hour == 16 and et_now.minute < 30): + return + + today = et_now.date() + + # Idempotency: already captured today? + existing = await pool.fetchval( + "SELECT 1 FROM portfolio_snapshots WHERE snapshot_date = $1 LIMIT 1", + today, + ) + if existing: + return + + # Portfolio snapshot from positions + positions: list = [] + portfolio_value = 0.0 + unrealized_pnl = 0.0 + try: + positions = await pool.fetch("SELECT * FROM positions WHERE quantity > 0") + portfolio_value = sum( + float(r["current_price"] or 0) * float(r["quantity"]) + for r in positions + ) + unrealized_pnl = sum(float(r["unrealized_pnl"] or 0) for r in positions) + + await pool.execute( + """INSERT INTO portfolio_snapshots + (snapshot_date, portfolio_value, unrealized_pnl, positions) + VALUES ($1, $2, $3, $4::jsonb)""", + today, + portfolio_value, + unrealized_pnl, + json.dumps([dict(r) for r in positions], default=str), + ) + logger.info("Captured portfolio snapshot: value=%.2f", portfolio_value) + except Exception: + logger.exception("Failed to capture portfolio snapshot") + + # Risk snapshot from daily activity + try: + daily_orders = await pool.fetchval( + "SELECT count(*) FROM orders WHERE created_at::date = $1", today + ) + daily_pnl = unrealized_pnl + + await pool.execute( + """INSERT INTO daily_risk_snapshots + (account_id, snapshot_date, portfolio_value, daily_pnl, daily_trade_count) + VALUES ((SELECT id FROM broker_accounts LIMIT 1), $1, $2, $3, $4) + ON CONFLICT DO NOTHING""", + today, + portfolio_value, + daily_pnl, + daily_orders or 0, + ) + logger.info("Captured risk snapshot: pnl=%.2f trades=%d", daily_pnl, daily_orders or 0) + except Exception: + logger.exception("Failed to capture risk snapshot") + + async def main() -> None: config = load_config() setup_logging("scheduler", level=config.log_level, json_output=config.json_logs) @@ -705,6 +802,8 @@ async def main() -> None: aggregation_counter = 0 report_consumer_counter = 0 report_schedule_counter = 0 + snapshot_counter = 0 + validation_counter = 0 try: while True: try: @@ -747,6 +846,16 @@ async def main() -> None: if report_schedule_counter >= REPORT_SCHEDULE_CYCLE_INTERVAL: report_schedule_counter = 0 await check_report_schedule(rds) + # Validation cycle: outcome evaluation + metric snapshots (~60 minutes) + validation_counter += 1 + if validation_counter >= VALIDATION_CYCLE_INTERVAL: + validation_counter = 0 + await run_validation_cycle(pool) + # Snapshot cycle: portfolio + risk snapshots daily after market close + snapshot_counter += 1 + if snapshot_counter >= SNAPSHOT_CYCLE_INTERVAL: + snapshot_counter = 0 + await maybe_capture_daily_snapshots(pool) finally: await release_lock(rds, "scheduler_cycle") except Exception: diff --git a/services/validation/prediction_snapshot.py b/services/validation/prediction_snapshot.py index f514f80..8908e39 100644 --- a/services/validation/prediction_snapshot.py +++ b/services/validation/prediction_snapshot.py @@ -311,7 +311,19 @@ async def create_prediction_snapshot( # 1. Fetch prices — handle NULL gracefully (Requirement 1.5) ticker_price = await fetch_latest_close_price(pool, ticker) if ticker_price is None: - logger.warning("No market price available for %s at snapshot time", ticker) + # Fallback: try positions table for tickers we actively hold (Bug 1.8) + pos_row = await pool.fetchrow( + "SELECT current_price FROM positions " + "WHERE ticker = $1 AND current_price IS NOT NULL LIMIT 1", + ticker, + ) + if pos_row: + ticker_price = float(pos_row["current_price"]) + logger.info( + "Used positions fallback price for %s: %s", ticker, ticker_price + ) + else: + logger.warning("No market price available for %s at snapshot time", ticker) spy_price = await fetch_latest_close_price(pool, "SPY") if spy_price is None: diff --git a/tests/test_scheduler_unit.py b/tests/test_scheduler_unit.py index ee7dfbb..8ed35f7 100644 --- a/tests/test_scheduler_unit.py +++ b/tests/test_scheduler_unit.py @@ -710,18 +710,20 @@ class TestCheckRateLimitEdgeCases: async def test_exactly_at_per_type_limit_allowed(self): """Count exactly equal to the limit should be allowed (only > limit blocks).""" rds = _mock_redis() - limit = DEFAULT_RATE_LIMITS["news_api"] + # Use filings_api (not in POLYGON_SOURCE_TYPES) to isolate the per-type check + limit = DEFAULT_RATE_LIMITS["filings_api"] rds.incr = AsyncMock(return_value=limit) - result = await check_rate_limit(rds, "news_api", _now()) + result = await check_rate_limit(rds, "filings_api", _now()) assert result is True @pytest.mark.asyncio async def test_one_over_per_type_limit_blocked(self): """Count one over the limit should be blocked.""" rds = _mock_redis() - limit = DEFAULT_RATE_LIMITS["news_api"] + # Use filings_api (not in POLYGON_SOURCE_TYPES) to isolate the per-type check + limit = DEFAULT_RATE_LIMITS["filings_api"] rds.incr = AsyncMock(return_value=limit + 1) - result = await check_rate_limit(rds, "news_api", _now()) + result = await check_rate_limit(rds, "filings_api", _now()) assert result is False @pytest.mark.asyncio