# Technical Design: ops-pipeline-fixes

## Overview

This design addresses 10 operational bugs that prevent the validation/calibration feedback loop from functioning and degrade ingestion throughput in the `stonks-beta` namespace. The fixes span the scheduler (rate limiting + periodic tasks), aggregation worker (config query), broker service (rejection reason), prediction snapshot (price fallback), and Helm chart (dead pod removal). All changes are localized with graceful fallbacks and no schema migrations required.

## Bug Details

Multiple operational bugs in the `stonks-beta` namespace prevent the validation/calibration feedback loop from functioning and degrade ingestion throughput. The core pipeline (ingestion → extraction → aggregation → recommendation → trading) flows end-to-end, but:
- The outcome evaluation → metrics computation → quality gate feedback loop is completely disconnected
- Polygon API rate limiting causes ~40% ingestion failures per cycle
- A broken config query prevents the v3 engine toggle from working
- Portfolio/risk snapshots are never captured
- Order rejection reasons are lost

```mermaid
graph TD
    subgraph "Scheduler (services/scheduler/app.py)"
        A[schedule_cycle] -->|paced enqueue| B[Ingestion Queue]
        C[validation_cycle] -->|hourly| D[evaluate_matured_predictions]
        C -->|after outcomes| E[compute_and_store_metric_snapshots]
        F[snapshot_cycle] -->|daily 16:30 ET| G[capture_portfolio_snapshot]
        F -->|daily 16:30 ET| H[capture_risk_snapshot]
    end

    subgraph "Aggregation (services/aggregation/worker.py)"
        I[_read_v3_flag] -->|fixed query| J[risk_configs.config JSONB]
    end

    subgraph "Broker (services/adapters/broker_service.py)"
        K[persist_order] -->|rejected status| L[orders.rejection_reason]
    end

    D --> O[prediction_outcomes]
    E --> P[model_metric_snapshots]
    P --> Q[Quality Gate]
```

## Expected Behavior

2.1 The scheduler SHALL pace Polygon API requests within the free-tier limit (~5 req/min), achieving near-zero 429 responses per cycle.

2.2 The aggregation worker SHALL read `v3_engine_enabled` from the `risk_configs` JSONB `config` column (not non-existent `key`/`value` columns).

2.3 The scheduler SHALL call `evaluate_matured_predictions()` hourly to populate `prediction_outcomes`.

2.4 The scheduler SHALL call `compute_and_store_metric_snapshots()` after outcome evaluation to populate `model_metric_snapshots`.

2.5 The quality gate SHALL have recent metric data available once the validation cycle runs.

2.6 The scheduler SHALL capture daily portfolio snapshots to `portfolio_snapshots` after market close.

2.7 The scheduler SHALL capture daily risk snapshots to `daily_risk_snapshots` after market close.

2.8 Prediction snapshots SHALL fall back to positions table prices when market_snapshots data is unavailable.

2.9 The lake-publisher deployment SHALL be scaled to 0 (idle pod, wasted resources).

2.10 The broker service SHALL populate `rejection_reason` on orders when broker or risk engine rejects.

## Hypothesized Root Cause

### Bug 1.1 — Polygon Rate Limiting
`POLYGON_GLOBAL_RATE_LIMIT = 45` in `services/scheduler/app.py` is set for a paid Polygon plan but the deployed instance uses the free tier (5 req/min). All 50+ sources are attempted per cycle, exhausting the limit instantly.

### Bug 1.2 — v3_engine_enabled Config Read
`_V3_ENGINE_FLAG_QUERY` in `services/aggregation/worker.py` reads `SELECT value FROM risk_configs WHERE key = 'v3_engine_enabled'`. The actual table has columns `name` (varchar) and `config` (JSONB) — no `key` or `value` column exists.

### Bug 1.3 & 1.4 — Outcome Evaluator & Metrics Never Scheduled
`evaluate_matured_predictions()` and `compute_and_store_metric_snapshots()` exist in `services/validation/` but are only imported in `services/trading/backtest_replay.py`. The scheduler main loop in `services/scheduler/app.py` has no call to either function.

### Bug 1.5 — Quality Gate Permanently Failing
`services/trading/model_quality_gate.py` queries `model_metric_snapshots` which is always empty (consequence of 1.4). Returns "no model metric snapshot available — defaulting to paper-only" every time.

### Bug 1.6 & 1.7 — Portfolio/Risk Snapshots
The trading engine has `_persist_daily_snapshot()` but it only executes when the engine's main loop is actively processing trades. The trading-engine pod shows only health checks — its main loop isn't cycling because there are no active trade triggers flowing through it. No fallback capture exists in the scheduler.

### Bug 1.8 — Market Price Gaps
`services/validation/prediction_snapshot.py` queries `market_snapshots` for price at prediction time. When Polygon rate limiting prevents market data fetches, no snapshot exists and `price_at_prediction` is NULL. 21% of snapshots affected.

### Bug 1.9 — Lake Publisher Idle
The standalone `lake-publisher` deployment polls `stonks:beta:queue:lake_publish` but all services (broker-adapter, recommendation) import `services.lake_publisher.worker` directly and publish inline — never pushing to the Redis queue.

### Bug 1.10 — Order rejection_reason NULL
`_INSERT_ORDER` SQL in `services/adapters/broker_service.py` doesn't include `rejection_reason` or `rejected_at` columns. The error is stored in `decision_trace` JSONB but the dedicated column stays NULL. The reconciliation path (`_reconcile_open_orders`) does set these columns, but initial persist does not.

## Fix Implementation

### Fix 1: Polygon Rate Limit Constant (Bug 1.1)

**File:** `services/scheduler/app.py`

Replace the hardcoded constant with an env-configurable value defaulting to 5:

```python
# Before:
POLYGON_GLOBAL_RATE_LIMIT: int = 45

# After:
POLYGON_GLOBAL_RATE_LIMIT: int = int(os.getenv("POLYGON_GLOBAL_RATE_LIMIT", "5"))
```

The existing `check_rate_limit()` function already implements per-minute windowed counting and skips sources once the limit is hit. By reducing the constant to match the free-tier limit, the system will naturally pace — enqueuing ~5 Polygon sources per minute across scheduler ticks (15s interval = 4 ticks/min). Skipped sources are retried next cycle.

**Validates:** Bugfix 2.1; Regression 3.1, 3.7

---

### Fix 2: v3_engine_enabled Config Query (Bug 1.2)

**File:** `services/aggregation/worker.py`

Replace the broken query and function:

```python
# Before:
_V3_ENGINE_FLAG_QUERY = """
SELECT value FROM risk_configs WHERE key = 'v3_engine_enabled'
"""

# After:
_V3_ENGINE_FLAG_QUERY = """
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 JSONB. Default False on error."""
    try:
        row = await pool.fetchrow(_V3_ENGINE_FLAG_QUERY)
        if row and row["enabled"]:
            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)
        return False
```

Reads from the `default` active risk_config's JSONB `config` field. Falls back to False (unchanged fail-safe).

**Validates:** Bugfix 2.2; Regression 3.2

---

### Fix 3: Validation Cycle in Scheduler (Bugs 1.3, 1.4, 1.5)

**File:** `services/scheduler/app.py`

Add a new periodic task (every ~240 ticks = ~60 minutes):

```python
# New constant:
VALIDATION_CYCLE_INTERVAL = int(os.getenv("VALIDATION_CYCLE_INTERVAL", "240"))

# New counter in main():
validation_counter = 0

# In main loop after existing periodic tasks:
validation_counter += 1
if validation_counter >= VALIDATION_CYCLE_INTERVAL:
    validation_counter = 0
    await run_validation_cycle(pool)
```

New function:

```python
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.outcome_evaluator import evaluate_matured_predictions
    from services.validation.metrics import compute_and_store_metric_snapshots

    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")
```

**Validates:** Bugfix 2.3, 2.4, 2.5; Regression 3.3

---

### Fix 4: Daily Portfolio & Risk Snapshots (Bugs 1.6, 1.7)

**File:** `services/scheduler/app.py`

Add a daily snapshot task that runs every ~60 minutes but only captures once per day after 16:30 ET:

```python
SNAPSHOT_CYCLE_INTERVAL = int(os.getenv("SNAPSHOT_CYCLE_INTERVAL", "240"))
snapshot_counter = 0

# In main loop:
snapshot_counter += 1
if snapshot_counter >= SNAPSHOT_CYCLE_INTERVAL:
    snapshot_counter = 0
    await maybe_capture_daily_snapshots(pool)
```

New function:

```python
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()

    # 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 + account data
    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 = sum(float(r["unrealized_pnl"] or 0) for r in positions) if positions else 0.0

        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")
```

**Validates:** Bugfix 2.6, 2.7; Regression 3.10

---

### Fix 5: Prediction Price Fallback (Bug 1.8)

**File:** `services/validation/prediction_snapshot.py`

After the primary `market_snapshots` price lookup returns NULL, add a fallback:

```python
# After market_snapshots lookup:
if price_at_prediction is None:
    pos_row = await conn.fetchrow(
        "SELECT current_price FROM positions "
        "WHERE ticker = $1 AND current_price IS NOT NULL LIMIT 1",
        ticker,
    )
    if pos_row:
        price_at_prediction = float(pos_row["current_price"])
```

Only covers tickers with open positions (currently 10). Acceptable tradeoff — most active tickers are the ones we hold.

**Validates:** Bugfix 2.8; Regression 3.4

---

### Fix 6: Lake Publisher Scale-Down (Bug 1.9)

**File:** `infra/helm/stonks-oracle/values.yaml`

```yaml
# Change:
    replicas: 0
```

Keeps the deployment definition intact for future use but schedules no pods.

**Validates:** Bugfix 2.9; Regression 3.5

---

### Fix 7: Order rejection_reason Population (Bug 1.10)

**File:** `services/adapters/broker_service.py`

Extend `_INSERT_ORDER` to include `rejection_reason` and `rejected_at`:

```python
_INSERT_ORDER = """
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, 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, $18, $19
)
ON CONFLICT (idempotency_key) DO UPDATE SET
    status = EXCLUDED.status,
    broker_order_id = EXCLUDED.broker_order_id,
    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()
"""
```

Update `persist_order()` to pass the new parameters:

```python
rejection_reason = resp.error if resp.status == OrderStatus.REJECTED else None
rejected_at = now if resp.status == OrderStatus.REJECTED else None
# Add as params $18, $19
```

**Validates:** Bugfix 2.10; Regression 3.6, 3.8

---

## Correctness Properties

Property 1: Rate limit compliance — After fix, the rolling 1-minute window for Polygon requests SHALL NOT exceed the configured limit (default 5). Existing `check_rate_limit()` windowed counter enforces this; we only change the threshold constant.

Property 2: Validation cycle completeness — `prediction_outcomes` row count SHALL grow monotonically after the first validation cycle runs. Each run finds matured snapshots not yet evaluated and persists outcomes.

Property 3: Metric snapshot freshness — `model_metric_snapshots` SHALL contain rows with `generated_at` within the last 2 hours after 2+ validation cycles. The quality gate can then evaluate against real data.

Property 4: Config read correctness — `_read_v3_flag()` SHALL return True when `risk_configs.config->>'v3_engine_enabled'` is `'true'` and False for all other values including NULL or missing key.

Property 5: Snapshot idempotency — `portfolio_snapshots` SHALL contain at most 1 row per `snapshot_date`. The `maybe_capture_daily_snapshots` function checks for existing rows before insert.

Property 6: Rejection reason preservation — Every order with `status = 'rejected'` persisted via `persist_order()` SHALL have a non-NULL `rejection_reason` extracted from the error response.

## Testing Strategy

- **Unit tests:** Update `test_scheduler.py` with a test verifying `run_validation_cycle` is called after the counter threshold. Test `_read_v3_flag` with mocked JSONB config returning various values.
- **Integration tests:** Verify `persist_order` with rejected status populates `rejection_reason` column.
- **Manual verification post-deploy:**
  - `kubectl logs deployment/scheduler -n stonks-beta --tail=100 | grep Validation` shows outcome counts
  - `SELECT count(*) FROM prediction_outcomes` starts growing within 1 hour
  - `SELECT count(*) FROM model_metric_snapshots` populates after outcomes exist
  - Scheduler logs show significantly fewer "Rate limit hit" warnings
  - Aggregation logs no longer show "column value does not exist" error
  - After market close: `SELECT * FROM portfolio_snapshots WHERE snapshot_date = CURRENT_DATE` returns 1 row

## Glossary

| Term | Definition |
|------|-----------|
| Validation cycle | Hourly scheduler task: evaluate_matured_predictions → compute_and_store_metric_snapshots |
| Quality gate | Threshold check on model_metric_snapshots that determines if trading can be promoted from paper to live |
| Prediction snapshot | Frozen state of a recommendation at generation time (prices, evidence, scores) |
| Outcome evaluation | Matching a matured prediction snapshot against realized market returns |
| Polygon free tier | API plan with ~5 requests/minute rate limit |