institutional-trader/IMPLEMENTATION_ROADMAP.md

Implementation Roadmap - Quick Reference

File-by-File Implementation Guide

Phase 1: Critical Features (Start Here)

1. Price Reaction Tracking

New File: backend/python_service/services/price_reaction_tracker.py

• Class: PriceReactionTracker
• Method: track_reaction(flow_row, pool) → returns dict with 5m/15m/30m reactions
• Integration: Call in main.py after enrich_flow_with_prices()

Modify: backend/python_service/main.py

# After line 145 (after price enrichment):
from services.price_reaction_tracker import PriceReactionTracker

reaction_tracker = PriceReactionTracker()
df_final = await reaction_tracker.enrich_with_reactions(df_final, pool)

---

2. VWAP Integration

Modify: backend/python_service/services/price_context.py

• Add method: async def calculate_vwap_at_time(symbol, timestamp, pool)
• Add method: async def get_vwap_for_date(symbol, date, pool)
• Integration: Call in enrich_flow_with_prices() method

Add to enrichment:

# In enrich_flow_with_prices(), add:
vwap_data = await self.get_vwap_at_time(symbol, flow_ts_utc, pool)
df['vwap_at_signal'] = vwap_data['vwap']
df['price_vs_vwap_pct'] = ((df['u_close'] - df['vwap_at_signal']) / df['vwap_at_signal']) * 100

---

3. Signal Tier Classification

New File: backend/python_service/services/signal_tier_classifier.py

• Class: SignalTierClassifier
• Method: classify_tier(row) → returns 'TIER_1', 'TIER_2', or 'IGNORE'

Modify: backend/python_service/services/options_flow_processor.py

• Add method: process_tier_classification(df) → adds signal_tier column
• Call in process() method after process_badges()

Integration:

# In process() method, after process_badges():
df = self.process_tier_classification(df)

---

4. Trade Checklist

New File: backend/python_service/services/trade_checklist.py

• Class: TradeChecklist
• Method: evaluate(flow_row) → returns checklist score and details

Modify: backend/python_service/main.py

• After all enrichments, add checklist evaluation:

from services.trade_checklist import TradeChecklist

checklist = TradeChecklist()
df_final['checklist_result'] = df_final.apply(
    lambda row: checklist.evaluate(row), axis=1
)
df_final['checklist_score'] = df_final['checklist_result'].apply(lambda x: x['checklist_score'])
df_final['checklist_passed'] = df_final['checklist_result'].apply(lambda x: x['checklist_passed'])

---

Phase 2: High Value Features

5. Strike Clustering

New File: backend/python_service/services/strike_cluster_detector.py

• Class: StrikeClusterDetector
• Method: detect_clusters(df, window_minutes=30) → adds cluster flags

Integration: Call in main.py after aggregations

---

6. Delta Weighting

Modify: backend/python_service/services/options_flow_processor.py

• Add method: calculate_delta_weighted_value(row)
• Add to process_aggregations() or create new method process_delta_weighting()

---

7. Index Correlation

New File: backend/python_service/services/index_correlation.py

• Class: IndexCorrelationService
• Method: check_index_alignment(flow_row, pool) → returns alignment data

Integration: Call in main.py after price enrichment

---

Phase 3: Advanced Features

8. Gamma Exposure

New File: backend/python_service/services/gamma_calculator.py

• Class: GammaCalculator
• Method: calculate_gex(df) → adds GEX columns

Note: Requires options pricing library (e.g., py_vollib or simplified approximation)

---

9. Sweep vs Block Detection

New File: backend/python_service/services/trade_type_detector.py

• Class: TradeTypeDetector
• Method: detect_trade_type(df) → adds trade_type column

---

10. DTE Buckets

Modify: backend/python_service/services/options_flow_processor.py

• Add method: calculate_dte_bucket(row)
• Add to process_moneyness() or create new method

---

Phase 4: Analytics

11. Historical Win Rate

New File: backend/python_service/services/pattern_analyzer.py

• Class: PatternAnalyzer
• Methods: track_pattern(), get_pattern_stats()

Database: Create table signal_patterns_history

---

12. Enhanced Entry/Exit Logic

Modify: backend/src/services/tradePlanGenerator.js

• Enhance generateEntryStrategy() with VWAP logic
• Enhance generateExitStrategy() with flow-based exits

---

Database Migrations

Migration 1: Add Enrichment Columns

ALTER TABLE processed_options_flow ADD COLUMN IF NOT EXISTS
    signal_tier VARCHAR(10),
    is_tradeable BOOLEAN,
    vwap_at_signal NUMERIC,
    price_vs_vwap_pct NUMERIC,
    price_reaction_5m_pct NUMERIC,
    price_reaction_15m_pct NUMERIC,
    flow_led_to_move BOOLEAN,
    checklist_score INTEGER,
    checklist_passed BOOLEAN;

Migration 2: Add Pattern Tracking Table

CREATE TABLE IF NOT EXISTS signal_patterns_history (
    id SERIAL PRIMARY KEY,
    pattern_hash VARCHAR(100),
    signal_time TIMESTAMPTZ,
    symbol VARCHAR(10),
    price_at_signal NUMERIC,
    price_5m_after NUMERIC,
    price_15m_after NUMERIC,
    outcome VARCHAR(20),
    return_pct NUMERIC,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_pattern_hash ON signal_patterns_history(pattern_hash);
CREATE INDEX idx_signal_time ON signal_patterns_history(signal_time);

---

API Endpoint Additions

Modify: backend/python_service/main.py

Add new endpoints:

@app.get("/api/options-flow/enhanced")
async def get_enhanced_flow(...):
    # Same as existing endpoint but with all enrichments enabled
    pass

@app.get("/api/options-flow/tier-1")
async def get_tier1_signals(...):
    # Filter to only Tier-1 signals
    df_final = df_final[df_final['signal_tier'] == 'TIER_1']
    pass

@app.get("/api/options-flow/checklist-passed")
async def get_checklist_passed(...):
    # Filter to only checklist-passed signals
    df_final = df_final[df_final['checklist_passed'] == True]
    pass

---

Testing Checklist

Unit Tests to Add

1. Price Reaction Tests

   • test_price_reaction_5m_positive()
   • test_price_reaction_no_move()
   • test_flow_led_to_move_detection()

2. Tier Classification Tests

   • test_tier1_classification()
   • test_tier2_classification()
   • test_ignore_classification()

3. Checklist Tests

   • test_checklist_5_5_passes()
   • test_checklist_4_5_passes()
   • test_checklist_3_5_fails()

4. VWAP Tests

   • test_vwap_calculation()
   • test_vwap_pullback_detection()
   • test_vwap_reclaim_detection()

---

Performance Considerations

Optimization Tips

1. Price Reaction Tracking

   • Batch fetch prices for all signals at once
   • Use async queries with connection pooling
   • Cache VWAP calculations per symbol/date

2. Strike Clustering

   • Use pandas groupby operations (already efficient)
   • Consider windowing for large datasets

3. Index Correlation

   • Cache SPY/QQQ flow data (update every minute)
   • Use materialized views for index flow aggregations

4. Gamma Calculation

   • Use simplified approximation (no full Black-Scholes)
   • Pre-calculate common strikes

---

Rollout Strategy

Week 1: Phase 1 (Critical)

• Day 1-2: Price Reaction Tracking
• Day 3-4: VWAP Integration
• Day 5: Signal Tier Classification
• Day 6-7: Trade Checklist

Week 2: Phase 2 (High Value)

• Day 1-2: Strike Clustering
• Day 3: Delta Weighting
• Day 4-5: Index Correlation

Week 3: Phase 3 (Advanced)

• Day 1-2: Gamma Exposure
• Day 3: Sweep vs Block
• Day 4: DTE Buckets

Week 4: Phase 4 (Analytics)

• Day 1-3: Historical Win Rate Tracking
• Day 4-5: Enhanced Entry/Exit Logic
• Day 6-7: Testing & Refinement

---

Monitoring & Metrics

Key Metrics to Track

1. Signal Quality

   • Tier-1 signal percentage
   • Checklist pass rate
   • Price reaction success rate

2. Trade Performance

   • Win rate by tier
   • Win rate by checklist score
   • Average return by pattern

3. System Performance

   • Enrichment processing time
   • Database query performance
   • API response times

---

Next Steps

1. ✅ Review this roadmap
2. ✅ Prioritize features based on your needs
3. ✅ Start with Phase 1 (Price Reaction + VWAP + Tier + Checklist)
4. ✅ Test each feature before moving to next
5. ✅ Monitor metrics and refine

---

Remember: Don't change existing code - extend it with new services and enrichments!