From b1cee2aa4358ef0abb0a95c7aaa56f2f303f2607 Mon Sep 17 00:00:00 2001 From: Deep Koluguri Date: Sun, 14 Dec 2025 20:42:35 -0500 Subject: [PATCH] added 5 day volume --- IMPLEMENTATION_ROADMAP.md | 317 +++++++ .../AI_ANALYSIS_SETUP.md | 0 {backend => README}/BLACKBOX_SYNC_GUIDE.md | 0 {backend => README}/CACHE_USAGE_EXAMPLE.md | 0 {backend => README}/CHEDDARFLOW_SYNC_GUIDE.md | 0 .../COMPLETION_STATUS.md | 0 {backend => README}/CONNECTION_STRING_FIX.md | 0 .../CORE_LOGIC_AND_UI_IMPROVEMENTS.md | 0 {backend => README}/CSV_IMPORT_GUIDE.md | 0 {backend => README}/DATABASE_QUOTA_FIX.md | 0 {backend => README}/DATABASE_SETUP.md | 0 {backend => README}/DATA_IMPORT_SUCCESS.md | 0 DEPLOYMENT.md => README/DEPLOYMENT.md | 0 .../DEPLOYMENT_QUICK_START.md | 0 .../DEPLOYMENT_SUMMARY.md | 0 .../FULL_STACK_ENHANCEMENT_ANALYSIS.md | 0 {backend => README}/HYBRID_SETUP.md | 0 .../IMPLEMENTATION_GUIDE.md | 0 .../IMPLEMENTATION_STATUS.md | 0 {backend => README}/INTEGRATION_COMPLETE.md | 0 {backend => README}/LOCAL_DB_SETUP.md | 0 .../MIGRATION_NOTES.md | 0 .../MOMENTUM_AND_MONEYNESS_IMPLEMENTED.md | 0 .../PROXMOX_DEPLOYMENT.md | 105 +++ .../QUERY_ENHANCEMENT_ANALYSIS.md | 0 .../QUERY_ENHANCEMENT_QUICKSTART.md | 0 {backend => README}/QUICK_FIX_DATABASE.md | 0 .../QUICK_INTEGRATION_GUIDE.md | 0 {backend => README}/QUICK_START.md | 0 {backend => README}/QUICK_START_BLACKBOX.md | 0 {backend => README}/QUICK_START_LOCAL.md | 0 {backend => README}/RAW_SQL_SETUP.md | 0 {backend => README}/SETUP_ENV.md | 0 .../START_SERVICE.md | 0 {backend => README}/TRADE_SIGNAL_FEATURES.md | 0 {backend => README}/TROUBLESHOOTING.md | 0 {backend => README}/TROUBLESHOOTING_SQL.md | 0 .../UI_FINALIZATION_SUMMARY.md | 0 {backend => README}/railway.toml | 0 SUGGESTIONS_TRADING_PLAYBOOK.md | 863 ++++++++++++++++++ .../__pycache__/db.cpython-312.pyc | Bin 3964 -> 4859 bytes .../__pycache__/main.cpython-312.pyc | Bin 14120 -> 15156 bytes .../options_flow_processor.cpython-312.pyc | Bin 37936 -> 37936 bytes .../__pycache__/price_context.cpython-312.pyc | Bin 15327 -> 15327 bytes backend/src/queries/optionsFlowQuery.js | 3 +- backend/src/routes/optionsFlow.js | 162 +++- backend/src/routes/stockPrices.js | 94 ++ backend/src/server.js | 2 + backend/src/services/volumeHistoryService.js | 144 +++ backend/src/services/yahooFinanceService.js | 33 + frontend/src/App.jsx | 10 +- .../components/dashboard/FlowInfoPanel.jsx | 186 ++++ .../dashboard/OptionsFlowCardList.jsx | 69 ++ .../components/dashboard/OptionsFlowPanel.jsx | 159 +++- .../dashboard/PerformanceTrackingPanel.jsx | 33 +- frontend/src/hooks/useOptionsFlow.js | 3 + frontend/src/hooks/useStockPrices.js | 61 ++ watch_and_upload.py | 0 watch_and_upload_blackbox_postgres.py | 733 +++++++++++++++ 59 files changed, 2911 insertions(+), 66 deletions(-) create mode 100644 IMPLEMENTATION_ROADMAP.md rename AI_ANALYSIS_SETUP.md => README/AI_ANALYSIS_SETUP.md (100%) rename {backend => README}/BLACKBOX_SYNC_GUIDE.md (100%) rename {backend => README}/CACHE_USAGE_EXAMPLE.md (100%) rename {backend => README}/CHEDDARFLOW_SYNC_GUIDE.md (100%) rename {backend/python_service => README}/COMPLETION_STATUS.md (100%) rename {backend => README}/CONNECTION_STRING_FIX.md (100%) rename CORE_LOGIC_AND_UI_IMPROVEMENTS.md => README/CORE_LOGIC_AND_UI_IMPROVEMENTS.md (100%) rename {backend => README}/CSV_IMPORT_GUIDE.md (100%) rename {backend => README}/DATABASE_QUOTA_FIX.md (100%) rename {backend => README}/DATABASE_SETUP.md (100%) rename {backend => README}/DATA_IMPORT_SUCCESS.md (100%) rename DEPLOYMENT.md => README/DEPLOYMENT.md (100%) rename DEPLOYMENT_QUICK_START.md => README/DEPLOYMENT_QUICK_START.md (100%) rename DEPLOYMENT_SUMMARY.md => README/DEPLOYMENT_SUMMARY.md (100%) rename FULL_STACK_ENHANCEMENT_ANALYSIS.md => README/FULL_STACK_ENHANCEMENT_ANALYSIS.md (100%) rename {backend => README}/HYBRID_SETUP.md (100%) rename IMPLEMENTATION_GUIDE.md => README/IMPLEMENTATION_GUIDE.md (100%) rename IMPLEMENTATION_STATUS.md => README/IMPLEMENTATION_STATUS.md (100%) rename {backend => README}/INTEGRATION_COMPLETE.md (100%) rename {backend => README}/LOCAL_DB_SETUP.md (100%) rename {backend/python_service => README}/MIGRATION_NOTES.md (100%) rename MOMENTUM_AND_MONEYNESS_IMPLEMENTED.md => README/MOMENTUM_AND_MONEYNESS_IMPLEMENTED.md (100%) rename PROXMOX_DEPLOYMENT.md => README/PROXMOX_DEPLOYMENT.md (94%) rename QUERY_ENHANCEMENT_ANALYSIS.md => README/QUERY_ENHANCEMENT_ANALYSIS.md (100%) rename QUERY_ENHANCEMENT_QUICKSTART.md => README/QUERY_ENHANCEMENT_QUICKSTART.md (100%) rename {backend => README}/QUICK_FIX_DATABASE.md (100%) rename QUICK_INTEGRATION_GUIDE.md => README/QUICK_INTEGRATION_GUIDE.md (100%) rename {backend => README}/QUICK_START.md (100%) rename {backend => README}/QUICK_START_BLACKBOX.md (100%) rename {backend => README}/QUICK_START_LOCAL.md (100%) rename {backend => README}/RAW_SQL_SETUP.md (100%) rename {backend => README}/SETUP_ENV.md (100%) rename {backend/python_service => README}/START_SERVICE.md (100%) rename {backend => README}/TRADE_SIGNAL_FEATURES.md (100%) rename {backend => README}/TROUBLESHOOTING.md (100%) rename {backend => README}/TROUBLESHOOTING_SQL.md (100%) rename UI_FINALIZATION_SUMMARY.md => README/UI_FINALIZATION_SUMMARY.md (100%) rename {backend => README}/railway.toml (100%) create mode 100644 SUGGESTIONS_TRADING_PLAYBOOK.md create mode 100644 backend/src/routes/stockPrices.js create mode 100644 backend/src/services/volumeHistoryService.js create mode 100644 frontend/src/components/dashboard/FlowInfoPanel.jsx create mode 100644 frontend/src/hooks/useStockPrices.js create mode 100644 watch_and_upload.py create mode 100644 watch_and_upload_blackbox_postgres.py diff --git a/IMPLEMENTATION_ROADMAP.md b/IMPLEMENTATION_ROADMAP.md new file mode 100644 index 0000000..5a33133 --- /dev/null +++ b/IMPLEMENTATION_ROADMAP.md @@ -0,0 +1,317 @@ +# 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` +```python +# 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:** +```python +# 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:** +```python +# 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: +```python +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 +```sql +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 +```sql +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: + +```python +@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! + diff --git a/AI_ANALYSIS_SETUP.md b/README/AI_ANALYSIS_SETUP.md similarity index 100% rename from AI_ANALYSIS_SETUP.md rename to README/AI_ANALYSIS_SETUP.md diff --git a/backend/BLACKBOX_SYNC_GUIDE.md b/README/BLACKBOX_SYNC_GUIDE.md similarity index 100% rename from backend/BLACKBOX_SYNC_GUIDE.md rename to README/BLACKBOX_SYNC_GUIDE.md diff --git a/backend/CACHE_USAGE_EXAMPLE.md b/README/CACHE_USAGE_EXAMPLE.md similarity index 100% rename from backend/CACHE_USAGE_EXAMPLE.md rename to README/CACHE_USAGE_EXAMPLE.md diff --git a/backend/CHEDDARFLOW_SYNC_GUIDE.md b/README/CHEDDARFLOW_SYNC_GUIDE.md similarity index 100% rename from backend/CHEDDARFLOW_SYNC_GUIDE.md rename to README/CHEDDARFLOW_SYNC_GUIDE.md diff --git a/backend/python_service/COMPLETION_STATUS.md b/README/COMPLETION_STATUS.md similarity index 100% rename from backend/python_service/COMPLETION_STATUS.md rename to README/COMPLETION_STATUS.md diff --git a/backend/CONNECTION_STRING_FIX.md b/README/CONNECTION_STRING_FIX.md similarity index 100% rename from backend/CONNECTION_STRING_FIX.md rename to README/CONNECTION_STRING_FIX.md diff --git a/CORE_LOGIC_AND_UI_IMPROVEMENTS.md b/README/CORE_LOGIC_AND_UI_IMPROVEMENTS.md similarity index 100% rename from CORE_LOGIC_AND_UI_IMPROVEMENTS.md rename to README/CORE_LOGIC_AND_UI_IMPROVEMENTS.md diff --git a/backend/CSV_IMPORT_GUIDE.md b/README/CSV_IMPORT_GUIDE.md similarity index 100% rename from backend/CSV_IMPORT_GUIDE.md rename to README/CSV_IMPORT_GUIDE.md diff --git a/backend/DATABASE_QUOTA_FIX.md b/README/DATABASE_QUOTA_FIX.md similarity index 100% rename from backend/DATABASE_QUOTA_FIX.md rename to README/DATABASE_QUOTA_FIX.md diff --git a/backend/DATABASE_SETUP.md b/README/DATABASE_SETUP.md similarity index 100% rename from backend/DATABASE_SETUP.md rename to README/DATABASE_SETUP.md diff --git a/backend/DATA_IMPORT_SUCCESS.md b/README/DATA_IMPORT_SUCCESS.md similarity index 100% rename from backend/DATA_IMPORT_SUCCESS.md rename to README/DATA_IMPORT_SUCCESS.md diff --git a/DEPLOYMENT.md b/README/DEPLOYMENT.md similarity index 100% rename from DEPLOYMENT.md rename to README/DEPLOYMENT.md diff --git a/DEPLOYMENT_QUICK_START.md b/README/DEPLOYMENT_QUICK_START.md similarity index 100% rename from DEPLOYMENT_QUICK_START.md rename to README/DEPLOYMENT_QUICK_START.md diff --git a/DEPLOYMENT_SUMMARY.md b/README/DEPLOYMENT_SUMMARY.md similarity index 100% rename from DEPLOYMENT_SUMMARY.md rename to README/DEPLOYMENT_SUMMARY.md diff --git a/FULL_STACK_ENHANCEMENT_ANALYSIS.md b/README/FULL_STACK_ENHANCEMENT_ANALYSIS.md similarity index 100% rename from FULL_STACK_ENHANCEMENT_ANALYSIS.md rename to README/FULL_STACK_ENHANCEMENT_ANALYSIS.md diff --git a/backend/HYBRID_SETUP.md b/README/HYBRID_SETUP.md similarity index 100% rename from backend/HYBRID_SETUP.md rename to README/HYBRID_SETUP.md diff --git a/IMPLEMENTATION_GUIDE.md b/README/IMPLEMENTATION_GUIDE.md similarity index 100% rename from IMPLEMENTATION_GUIDE.md rename to README/IMPLEMENTATION_GUIDE.md diff --git a/IMPLEMENTATION_STATUS.md b/README/IMPLEMENTATION_STATUS.md similarity index 100% rename from IMPLEMENTATION_STATUS.md rename to README/IMPLEMENTATION_STATUS.md diff --git a/backend/INTEGRATION_COMPLETE.md b/README/INTEGRATION_COMPLETE.md similarity index 100% rename from backend/INTEGRATION_COMPLETE.md rename to README/INTEGRATION_COMPLETE.md diff --git a/backend/LOCAL_DB_SETUP.md b/README/LOCAL_DB_SETUP.md similarity index 100% rename from backend/LOCAL_DB_SETUP.md rename to README/LOCAL_DB_SETUP.md diff --git a/backend/python_service/MIGRATION_NOTES.md b/README/MIGRATION_NOTES.md similarity index 100% rename from backend/python_service/MIGRATION_NOTES.md rename to README/MIGRATION_NOTES.md diff --git a/MOMENTUM_AND_MONEYNESS_IMPLEMENTED.md b/README/MOMENTUM_AND_MONEYNESS_IMPLEMENTED.md similarity index 100% rename from MOMENTUM_AND_MONEYNESS_IMPLEMENTED.md rename to README/MOMENTUM_AND_MONEYNESS_IMPLEMENTED.md diff --git a/PROXMOX_DEPLOYMENT.md b/README/PROXMOX_DEPLOYMENT.md similarity index 94% rename from PROXMOX_DEPLOYMENT.md rename to README/PROXMOX_DEPLOYMENT.md index e8c3261..99d7679 100644 --- a/PROXMOX_DEPLOYMENT.md +++ b/README/PROXMOX_DEPLOYMENT.md @@ -1042,6 +1042,111 @@ sudo -u postgres psql institutional_trader < backup_20240101.sql ## 12. Troubleshooting +### Git Pull Fails - DNS Resolution Error + +**Error:** `fatal: unable to access 'https://github.com/...': Could not resolve host: github.com` + +**This means your container/VM cannot resolve DNS names.** + +**Quick Fix:** + +```bash +# 1. Check current DNS configuration +cat /etc/resolv.conf + +# 2. If empty or missing, add DNS servers +sudo nano /etc/resolv.conf +``` + +Add these lines: +``` +nameserver 8.8.8.8 +nameserver 8.8.4.4 +nameserver 1.1.1.1 +``` + +**For LXC Containers (Persistent Fix):** + +```bash +# Edit container DNS configuration +# On Proxmox host, edit container config: +nano /etc/pve/lxc/.conf + +# Add DNS servers: +nameserver: 8.8.8.8 +nameserver: 8.8.4.4 +``` + +Or via Proxmox Web UI: +- Go to your container → Options → DNS +- Set DNS servers: `8.8.8.8, 8.8.4.4` + +**For Ubuntu VMs/Containers (Systemd-resolved):** + +```bash +# Edit systemd-resolved config +sudo nano /etc/systemd/resolved.conf +``` + +Uncomment and set: +``` +[Resolve] +DNS=8.8.8.8 8.8.4.4 1.1.1.1 +FallbackDNS=1.1.1.1 8.8.8.8 +``` + +**If using Tailscale (DNS server 100.100.100.100):** + +If your `/etc/resolv.conf` shows Tailscale DNS (`100.100.100.100`) but it's timing out, add fallback DNS servers: + +```bash +# Edit systemd-resolved config +sudo nano /etc/systemd/resolved.conf +``` + +Set: +``` +[Resolve] +DNS=100.100.100.100 8.8.8.8 8.8.4.4 1.1.1.1 +FallbackDNS=8.8.8.8 8.8.4.4 1.1.1.1 +``` + +This keeps Tailscale DNS as primary (for Tailscale network access) but adds public DNS as fallback. + +```bash +# Restart systemd-resolved +sudo systemctl restart systemd-resolved + +# Test DNS +nslookup github.com +ping -c 2 github.com +``` + +**Alternative: Use IP Address (Temporary Workaround)** + +If DNS still doesn't work, you can manually update git remote: + +```bash +# Get GitHub IP +nslookup github.com 8.8.8.8 + +# Or use known GitHub IPs (may change) +# Update git remote to use IP (not recommended for long-term) +git remote set-url origin https://140.82.121.3/deepkoluguri/INSTITUTIONAL-FLOW-TRADING-PLATFORM.git +``` + +**Verify DNS is Working:** + +```bash +# Test DNS resolution +nslookup github.com +dig github.com + +# Test connectivity +ping -c 2 github.com +curl -I https://github.com +``` + ### Backend Won't Start ```bash diff --git a/QUERY_ENHANCEMENT_ANALYSIS.md b/README/QUERY_ENHANCEMENT_ANALYSIS.md similarity index 100% rename from QUERY_ENHANCEMENT_ANALYSIS.md rename to README/QUERY_ENHANCEMENT_ANALYSIS.md diff --git a/QUERY_ENHANCEMENT_QUICKSTART.md b/README/QUERY_ENHANCEMENT_QUICKSTART.md similarity index 100% rename from QUERY_ENHANCEMENT_QUICKSTART.md rename to README/QUERY_ENHANCEMENT_QUICKSTART.md diff --git a/backend/QUICK_FIX_DATABASE.md b/README/QUICK_FIX_DATABASE.md similarity index 100% rename from backend/QUICK_FIX_DATABASE.md rename to README/QUICK_FIX_DATABASE.md diff --git a/QUICK_INTEGRATION_GUIDE.md b/README/QUICK_INTEGRATION_GUIDE.md similarity index 100% rename from QUICK_INTEGRATION_GUIDE.md rename to README/QUICK_INTEGRATION_GUIDE.md diff --git a/backend/QUICK_START.md b/README/QUICK_START.md similarity index 100% rename from backend/QUICK_START.md rename to README/QUICK_START.md diff --git a/backend/QUICK_START_BLACKBOX.md b/README/QUICK_START_BLACKBOX.md similarity index 100% rename from backend/QUICK_START_BLACKBOX.md rename to README/QUICK_START_BLACKBOX.md diff --git a/backend/QUICK_START_LOCAL.md b/README/QUICK_START_LOCAL.md similarity index 100% rename from backend/QUICK_START_LOCAL.md rename to README/QUICK_START_LOCAL.md diff --git a/backend/RAW_SQL_SETUP.md b/README/RAW_SQL_SETUP.md similarity index 100% rename from backend/RAW_SQL_SETUP.md rename to README/RAW_SQL_SETUP.md diff --git a/backend/SETUP_ENV.md b/README/SETUP_ENV.md similarity index 100% rename from backend/SETUP_ENV.md rename to README/SETUP_ENV.md diff --git a/backend/python_service/START_SERVICE.md b/README/START_SERVICE.md similarity index 100% rename from backend/python_service/START_SERVICE.md rename to README/START_SERVICE.md diff --git a/backend/TRADE_SIGNAL_FEATURES.md b/README/TRADE_SIGNAL_FEATURES.md similarity index 100% rename from backend/TRADE_SIGNAL_FEATURES.md rename to README/TRADE_SIGNAL_FEATURES.md diff --git a/backend/TROUBLESHOOTING.md b/README/TROUBLESHOOTING.md similarity index 100% rename from backend/TROUBLESHOOTING.md rename to README/TROUBLESHOOTING.md diff --git a/backend/TROUBLESHOOTING_SQL.md b/README/TROUBLESHOOTING_SQL.md similarity index 100% rename from backend/TROUBLESHOOTING_SQL.md rename to README/TROUBLESHOOTING_SQL.md diff --git a/UI_FINALIZATION_SUMMARY.md b/README/UI_FINALIZATION_SUMMARY.md similarity index 100% rename from UI_FINALIZATION_SUMMARY.md rename to README/UI_FINALIZATION_SUMMARY.md diff --git a/backend/railway.toml b/README/railway.toml similarity index 100% rename from backend/railway.toml rename to README/railway.toml diff --git a/SUGGESTIONS_TRADING_PLAYBOOK.md b/SUGGESTIONS_TRADING_PLAYBOOK.md new file mode 100644 index 0000000..2636d2e --- /dev/null +++ b/SUGGESTIONS_TRADING_PLAYBOOK.md @@ -0,0 +1,863 @@ +# Trading Playbook Implementation Suggestions + +## Current State Analysis + +### ✅ What You Already Have +- Badge system: 🟢/🔴, 💎, ⭐, 💰, ⚡, 🚀 +- Rocket scoring algorithm +- Price context: RTH open, prior close, 5m/15m momentum +- Tape alignment detection +- Trade signal generation +- Session bucketing (PRE/RTH/POST) +- Premium filtering and aggregations + +### ❌ What's Missing (High Impact) + +--- + +## PART A — TRADING LOGIC ENHANCEMENTS + +### 1️⃣ Signal Tier Classification System + +**Current Gap:** All signals are treated equally. You need to classify them into Tier-1, Tier-2, and Ignore categories. + +**Suggestion:** +- **Add a new service:** `backend/python_service/services/signal_tier_classifier.py` + - Classify signals based on badge combinations + - Tier-1: 🟢/🔴 + 💎 + ⭐ + premium > 500K + direction aligned + - Tier-2: 🟢 + 💎 (no ⭐) OR ⭐ without 💎 + - Ignore: OTM-only, mixed signals, low volume/OI ratio + +**Implementation Approach:** +```python +# In options_flow_processor.py, add after process_badges(): +def classify_signal_tier(row): + badge_round = row.get('badge_round', '') + badge_more = row.get('badge_more', '') + premium = row.get('premium_num', 0) or 0 + direction = row.get('direction', '') + bull_total = row.get('bull_total', 0) or 0 + bear_total = row.get('bear_total', 0) or 0 + + has_diamond = '💎' in badge_more + has_star = '⭐' in badge_more + + # Tier-1 conditions + if (badge_round in ['🟢', '🔴'] and + has_diamond and has_star and + premium > 500000): + # Check direction alignment + if (badge_round == '🟢' and direction == 'BULL' and (bull_total - bear_total) > 0): + return 'TIER_1' + elif (badge_round == '🔴' and direction == 'BEAR' and (bear_total - bull_total) > 0): + return 'TIER_1' + + # Tier-2 conditions + if (badge_round == '🟢' and has_diamond and not has_star): + return 'TIER_2' + if (has_star and not has_diamond): + return 'TIER_2' + + # Ignore conditions + # (Add logic for OTM-only, mixed signals, etc.) + + return 'IGNORE' +``` + +**Database Addition:** +- Add `signal_tier` column to processed flow output +- Add `is_tradeable` boolean flag + +--- + +### 2️⃣ VWAP Integration + +**Current Gap:** You have price context but no VWAP calculation or VWAP-based entry/exit logic. + +**Suggestion:** +- **Extend:** `backend/python_service/services/price_context.py` + - Add `calculate_vwap()` method + - Calculate VWAP for each symbol on each trading day + - Store VWAP at signal time + - Calculate distance from VWAP (percentage) + +**Implementation Approach:** +```python +# Add to PriceContextService: +async def get_vwap_at_time(self, symbol: str, timestamp: datetime, pool: asyncpg.Pool): + """Calculate VWAP up to the given timestamp for the trading day""" + # Query all 1m bars from RTH open to timestamp + # Calculate: SUM(price * volume) / SUM(volume) + # Return VWAP value and distance from current price +``` + +**New Fields to Add:** +- `vwap_at_signal` - VWAP value at signal time +- `price_vs_vwap_pct` - Percentage distance from VWAP +- `vwap_reclaimed` - Boolean: did price reclaim VWAP after signal? + +**Entry Strategy Integration:** +- Best entry: VWAP pullback or VWAP reclaim +- Good entry: Break & hold above prior high +- Avoid: Chasing vertical candles + +--- + +### 3️⃣ Price Reaction Tracking (MOST IMPORTANT) + +**Current Gap:** No tracking of how price moves AFTER the signal appears. + +**Suggestion:** +- **New service:** `backend/python_service/services/price_reaction_tracker.py` + - Track price 5 minutes, 15 minutes, 30 minutes after signal + - Calculate price change percentage + - Identify if flow led to price movement or was just hedging + +**Implementation Approach:** +```python +class PriceReactionTracker: + async def track_reaction(self, flow_row, pool): + signal_time = flow_row['flow_ts_utc'] + symbol = flow_row['symbol_norm'] + price_at_signal = flow_row['u_close'] + + # Get price 5m, 15m, 30m after signal + price_5m = await get_price_at_time(symbol, signal_time + timedelta(minutes=5)) + price_15m = await get_price_at_time(symbol, signal_time + timedelta(minutes=15)) + price_30m = await get_price_at_time(symbol, signal_time + timedelta(minutes=30)) + + # Calculate reactions + reaction_5m = ((price_5m - price_at_signal) / price_at_signal) * 100 if price_5m else None + reaction_15m = ((price_15m - price_at_signal) / price_at_signal) * 100 if price_15m else None + reaction_30m = ((price_30m - price_at_signal) / price_at_signal) * 100 if price_30m else None + + # High/Low break confirmation + high_break = price_5m > flow_row.get('u_high', 0) + low_break = price_5m < flow_row.get('u_low', 0) + + return { + 'price_reaction_5m_pct': reaction_5m, + 'price_reaction_15m_pct': reaction_15m, + 'price_reaction_30m_pct': reaction_30m, + 'high_break_5m': high_break, + 'low_break_5m': low_break, + 'flow_led_to_move': reaction_5m and abs(reaction_5m) > 0.5 # 0.5% threshold + } +``` + +**Database Addition:** +- Add columns: `price_reaction_5m_pct`, `price_reaction_15m_pct`, `high_break_5m`, `low_break_5m` +- Add flag: `flow_led_to_move` (boolean) + +**Why This Matters:** +- Flow without price reaction = hedge or roll (ignore) +- Flow with price reaction = real positioning (trade it) + +--- + +### 4️⃣ Strike Clustering Detection + +**Current Gap:** No detection of multiple large trades at the same strike (institutional layering). + +**Suggestion:** +- **New service:** `backend/python_service/services/strike_cluster_detector.py` + - Group trades by strike and expiration + - Identify clusters: 3+ trades at same strike within 30 minutes + - Calculate cluster premium total + - Flag as "institutional positioning" vs "single trade" + +**Implementation Approach:** +```python +class StrikeClusterDetector: + def detect_clusters(self, df: pd.DataFrame, window_minutes: int = 30): + """Detect strike clusters within time window""" + df = df.copy() + + # Group by symbol, exp_date, strike + clusters = df.groupby(['symbol_norm', 'exp_date', 'strike_num']).apply( + lambda g: self._find_clusters_in_group(g, window_minutes) + ) + + return clusters + + def _find_clusters_in_group(self, group, window_minutes): + """Find time-based clusters within a strike group""" + # Sort by time + group = group.sort_values('flow_ts_utc') + + # Rolling window: if 3+ trades within window_minutes, it's a cluster + # Return cluster flags and cluster IDs +``` + +**New Fields:** +- `is_cluster_trade` - Boolean +- `cluster_id` - Unique ID for the cluster +- `cluster_size` - Number of trades in cluster +- `cluster_total_premium` - Sum of all premiums in cluster + +**Why This Matters:** +- Institutions rarely place one order — they layer +- Clusters = stronger signal than single prints + +--- + +### 5️⃣ Gamma Exposure (GEX) Calculation + +**Current Gap:** No gamma exposure tracking. This explains why some rockets fail. + +**Suggestion:** +- **New service:** `backend/python_service/services/gamma_calculator.py` + - Calculate call GEX and put GEX per strike + - Net dealer gamma = Call GEX - Put GEX + - Positive GEX = price pinned (resistance) + - Negative GEX = explosive moves possible + +**Implementation Approach:** +```python +class GammaCalculator: + def calculate_gex(self, df: pd.DataFrame): + """ + Calculate Gamma Exposure (GEX) + GEX = OI * Spot^2 * Gamma * 0.01 * Multiplier + Simplified: GEX ≈ OI * Spot^2 * 0.01 (for rough estimate) + """ + # For each strike, calculate: + # - Call GEX (positive for calls) + # - Put GEX (negative for puts) + # - Net GEX = Call GEX + Put GEX + + # Add to flow row: + # - strike_gex (GEX at this strike) + # - net_dealer_gex (aggregate GEX for symbol) + # - gex_pin_level (strike with highest GEX) +``` + +**New Fields:** +- `strike_gex` - GEX at this strike +- `net_dealer_gex` - Net GEX for the symbol +- `gex_pin_level` - Strike where GEX is highest (pin level) +- `is_gex_positive` - Boolean: positive GEX = pinning, negative = explosive + +**Why This Matters:** +- +GEX = Price pinned (rockets may fail at pin level) +- -GEX = Explosive moves (rockets more likely to work) + +--- + +### 6️⃣ Delta Weighting (Smart Money Filter) + +**Current Gap:** No delta weighting. ITM delta > OTM lottery tickets. + +**Suggestion:** +- **Extend:** `backend/python_service/services/options_flow_processor.py` + - Add delta calculation (approximate: use Black-Scholes or simplified formula) + - Calculate: `delta_weighted_premium = delta * volume * premium` + - Filter out low delta-weighted trades (YOLO prints) + +**Implementation Approach:** +```python +def calculate_delta_weighted_value(row): + """Calculate delta-weighted premium value""" + # Simplified delta approximation: + # For CALL: delta ≈ N(d1) where d1 = (ln(S/K) + (r+σ²/2)*T) / (σ*√T) + # For rough estimate: delta ≈ 0.5 for ATM, 0.8+ for ITM, 0.2- for OTM + + spot = row.get('spot_num', 0) + strike = row.get('strike_num', 0) + cp = row.get('cp_norm', '') + moneyness = row.get('moneyness', '') + + # Simplified delta based on moneyness + if moneyness == 'ITM': + delta = 0.7 if cp == 'CALL' else 0.7 + elif moneyness == 'OTM': + delta = 0.3 if cp == 'CALL' else 0.3 + else: # ATM + delta = 0.5 + + volume = row.get('vol_num', 0) or 0 + premium = row.get('premium_num', 0) or 0 + + return delta * volume * premium +``` + +**New Fields:** +- `delta_approx` - Approximate delta value +- `delta_weighted_premium` - Delta * Volume * Premium +- `is_smart_money` - Boolean: delta_weighted_premium > threshold + +**Why This Matters:** +- Filters out YOLO OTM lottery prints +- ITM delta > OTM = real positioning + +--- + +### 7️⃣ Time-to-Expiration Buckets + +**Current Gap:** No DTE-based classification. + +**Suggestion:** +- **Extend:** `backend/python_service/services/options_flow_processor.py` + - Calculate DTE (days to expiration) + - Bucket into: 0DTE, 1-3 DTE, 7-14 DTE, Monthly + - Different logic per bucket + +**Implementation Approach:** +```python +def calculate_dte_bucket(row): + """Calculate days to expiration and bucket""" + exp_date = row.get('exp_date') + flow_date = row.get('flow_date_cst') + + if not exp_date or not flow_date: + return None + + if isinstance(flow_date, datetime): + flow_date = flow_date.date() + if isinstance(exp_date, datetime): + exp_date = exp_date.date() + + dte = (exp_date - flow_date).days + + if dte == 0: + return '0DTE' + elif 1 <= dte <= 3: + return '1-3DTE' + elif 4 <= dte <= 6: + return '4-6DTE' + elif 7 <= dte <= 14: + return '7-14DTE' + elif 15 <= dte <= 30: + return 'MONTHLY' + else: + return 'LONG_TERM' +``` + +**New Fields:** +- `dte` - Days to expiration +- `dte_bucket` - Bucket classification +- `is_0dte` - Boolean flag + +**Why This Matters:** +- 0DTE → intraday pressure (gamma risk) +- Longer DTE → directional thesis (less gamma risk) + +--- + +### 8️⃣ Sweep vs Block Detection + +**Current Gap:** No distinction between sweeps (urgency) and blocks (positioning). + +**Suggestion:** +- **New service:** `backend/python_service/services/trade_type_detector.py` + - Detect multiple trades at same strike/expiration within 2 seconds = SWEEP + - Single large trade = BLOCK + - Different trading implications + +**Implementation Approach:** +```python +class TradeTypeDetector: + def detect_trade_type(self, df: pd.DataFrame): + """Detect if trade is sweep or block""" + df = df.copy() + df = df.sort_values(['symbol_norm', 'exp_date', 'strike_num', 'flow_ts_utc']) + + # Group by symbol, exp, strike + groups = df.groupby(['symbol_norm', 'exp_date', 'strike_num']) + + def classify_group(group): + # If multiple trades within 2 seconds = sweep + # If single large trade = block + # Otherwise = regular trade + + if len(group) == 1: + return 'BLOCK' if group.iloc[0]['premium_num'] > 500000 else 'REGULAR' + + # Check time differences + time_diffs = group['flow_ts_utc'].diff().dt.total_seconds() + has_sweep = (time_diffs <= 2).any() + + if has_sweep: + return 'SWEEP' + else: + return 'CLUSTER' + + df['trade_type'] = groups.apply(classify_group).values + + return df +``` + +**New Fields:** +- `trade_type` - 'SWEEP', 'BLOCK', 'CLUSTER', 'REGULAR' +- `is_sweep` - Boolean +- `is_block` - Boolean + +**Why This Matters:** +- Sweeps = urgency (institutions hitting multiple exchanges) +- Blocks = positioning (single large order) + +--- + +### 9️⃣ Historical Win Rate Tracking + +**Current Gap:** No tracking of which patterns actually work. + +**Suggestion:** +- **New service:** `backend/python_service/services/pattern_analyzer.py` + - Track pattern → outcome mapping + - Calculate win rate per pattern + - Average return per pattern + - Max drawdown per pattern + +**Database Addition:** +- **New table:** `signal_patterns_history` + - Columns: pattern_hash, signal_time, price_at_signal, price_5m_after, price_15m_after, outcome, return_pct + +**Implementation Approach:** +```python +class PatternAnalyzer: + def track_pattern(self, flow_row, price_reaction): + """Track pattern and outcome""" + pattern_hash = self._hash_pattern(flow_row) + + # Store in database: + # - Pattern signature (badge combo + premium tier + DTE) + # - Outcome (price reaction) + # - Return percentage + + def get_pattern_stats(self, pattern_hash): + """Get historical stats for a pattern""" + # Query database for all instances of this pattern + # Calculate: win_rate, avg_return, max_drawdown +``` + +**New Fields:** +- `pattern_hash` - Unique identifier for pattern +- `historical_win_rate` - Win rate for this pattern +- `historical_avg_return` - Average return for this pattern +- `pattern_confidence` - Confidence based on historical performance + +**Why This Matters:** +- Discover which patterns actually work +- 🚀🚀 without 💎 fails more often +- 🟢💎⭐ + VWAP reclaim wins most + +--- + +### 🔟 Index & Correlation Filter + +**Current Gap:** No SPY/QQQ/VIX alignment check. + +**Suggestion:** +- **New service:** `backend/python_service/services/index_correlation.py` + - Fetch SPY/QQQ flow at signal time + - Check VIX direction + - Rule: Single stock flow works best when index agrees + +**Implementation Approach:** +```python +class IndexCorrelationService: + async def check_index_alignment(self, flow_row, pool): + """Check if index flow aligns with stock flow""" + symbol = flow_row['symbol_norm'] + signal_time = flow_row['flow_ts_utc'] + direction = flow_row['direction'] + + # Get SPY/QQQ flow in same time window + spy_flow = await self.get_index_flow('SPY', signal_time, pool) + qqq_flow = await self.get_index_flow('QQQ', signal_time, pool) + + # Get VIX direction + vix_direction = await self.get_vix_direction(signal_time, pool) + + # Check alignment + index_bullish = (spy_flow.get('net_premium', 0) > 0) or (qqq_flow.get('net_premium', 0) > 0) + index_bearish = (spy_flow.get('net_premium', 0) < 0) or (qqq_flow.get('net_premium', 0) < 0) + + aligned = ( + (direction == 'BULL' and index_bullish) or + (direction == 'BEAR' and index_bearish) + ) + + return { + 'index_aligned': aligned, + 'spy_flow_direction': 'BULL' if spy_flow.get('net_premium', 0) > 0 else 'BEAR', + 'qqq_flow_direction': 'BULL' if qqq_flow.get('net_premium', 0) > 0 else 'BEAR', + 'vix_direction': vix_direction + } +``` + +**New Fields:** +- `index_aligned` - Boolean: does index flow agree? +- `spy_flow_direction` - SPY flow direction +- `qqq_flow_direction` - QQQ flow direction +- `vix_direction` - VIX direction (up/down) + +**Why This Matters:** +- Single stock flow works best when index agrees +- Contrarian flow (stock vs index) = lower probability + +--- + +## PART B — TRADE CHECKLIST IMPLEMENTATION + +### Trade Entry Checklist + +**Suggestion:** +- **New service:** `backend/python_service/services/trade_checklist.py` + - Implement 5-point checklist + - Return checklist score (0-5) + - Only allow trades with 4/5 or 5/5 + +**Implementation Approach:** +```python +class TradeChecklist: + def evaluate(self, flow_row): + """Evaluate trade checklist""" + checks = { + 'has_direction': flow_row.get('badge_round') in ['🟢', '🔴'], + 'has_diamond': '💎' in flow_row.get('badge_more', ''), + 'has_star': '⭐' in flow_row.get('badge_more', ''), + 'price_respects_vwap': self._check_vwap_respect(flow_row), + 'index_confirms': flow_row.get('index_aligned', False) + } + + score = sum(checks.values()) + passed = score >= 4 + + return { + 'checklist_score': score, + 'checklist_passed': passed, + 'checks': checks + } +``` + +**New Fields:** +- `checklist_score` - 0-5 score +- `checklist_passed` - Boolean: 4/5 or 5/5 +- `checklist_details` - JSON with individual check results + +--- + +## PART C — ENHANCED ENTRY/EXIT LOGIC + +### Entry Strategy Enhancement + +**Current Gap:** Entry logic exists but doesn't use VWAP pullback/reclaim. + +**Suggestion:** +- **Extend:** `backend/src/services/tradePlanGenerator.js` + - Add VWAP pullback entry + - Add VWAP reclaim entry + - Add prior high break entry + - Avoid chasing vertical candles + +**Implementation:** +```javascript +function generateEntryStrategy(signal, currentPrice, priceContext) { + const vwap = priceContext.vwap; + const priorHigh = priceContext.priorHigh; + const vwapDistance = ((currentPrice - vwap) / vwap) * 100; + + if (signal === 'BUY') { + // Best: VWAP pullback or VWAP reclaim + if (currentPrice < vwap && vwapDistance > -1) { + return { + type: 'VWAP_PULLBACK', + entry: vwap * 0.998, // Slightly below VWAP + reason: 'VWAP pullback entry' + }; + } + + // Good: Break & hold above prior high + if (currentPrice > priorHigh) { + return { + type: 'BREAKOUT', + entry: priorHigh * 1.001, // Slightly above prior high + reason: 'Prior high breakout' + }; + } + + // Avoid: Chasing vertical candles + if (vwapDistance > 2) { + return { + type: 'WAIT', + reason: 'Price too extended from VWAP - wait for pullback' + }; + } + } + // Similar for SELL signals... +} +``` + +--- + +### Exit Strategy Enhancement + +**Current Gap:** Exit logic is basic. Need flow-based exits. + +**Suggestion:** +- **Extend:** `backend/src/services/tradePlanGenerator.js` + - Exit when flow stalls + - Exit when opposite 💎 appears + - Exit when net premium flips + - Exit when price rejects VWAP + - Scale out at +30-50% option gain + +**Implementation:** +```javascript +function generateExitStrategy(signal, entryPrice, currentPrice, flowData) { + const exits = []; + + // Flow stalls + if (flowData.recentFlowVolume < flowData.avgFlowVolume * 0.3) { + exits.push({ + type: 'FLOW_STALL', + reason: 'Flow volume dropped significantly' + }); + } + + // Opposite diamond appears + if (signal === 'BUY' && flowData.hasBearDiamond) { + exits.push({ + type: 'OPPOSITE_SIGNAL', + reason: 'Bear diamond (💎) appeared - exit long' + }); + } + + // Net premium flips + if (signal === 'BUY' && flowData.netPremium < 0) { + exits.push({ + type: 'PREMIUM_FLIP', + reason: 'Net premium flipped negative' + }); + } + + // Price rejects VWAP + if (currentPrice < priceContext.vwap && signal === 'BUY') { + exits.push({ + type: 'VWAP_REJECTION', + reason: 'Price rejected VWAP - exit' + }); + } + + // Scale out at gains + const gainPct = ((currentPrice - entryPrice) / entryPrice) * 100; + if (gainPct >= 30) { + exits.push({ + type: 'SCALE_OUT', + reason: `+${gainPct.toFixed(1)}% gain - scale out 50%` + }); + } + + return exits; +} +``` + +--- + +## PART D — DATABASE SCHEMA ADDITIONS + +### New Columns for `processed_options_flow` (or new enrichment table) + +```sql +-- Signal classification +signal_tier VARCHAR(10), -- 'TIER_1', 'TIER_2', 'IGNORE' +is_tradeable BOOLEAN, + +-- VWAP +vwap_at_signal NUMERIC, +price_vs_vwap_pct NUMERIC, +vwap_reclaimed BOOLEAN, + +-- Price reaction +price_reaction_5m_pct NUMERIC, +price_reaction_15m_pct NUMERIC, +price_reaction_30m_pct NUMERIC, +high_break_5m BOOLEAN, +low_break_5m BOOLEAN, +flow_led_to_move BOOLEAN, + +-- Strike clustering +is_cluster_trade BOOLEAN, +cluster_id VARCHAR(50), +cluster_size INTEGER, +cluster_total_premium NUMERIC, + +-- Gamma exposure +strike_gex NUMERIC, +net_dealer_gex NUMERIC, +gex_pin_level NUMERIC, +is_gex_positive BOOLEAN, + +-- Delta weighting +delta_approx NUMERIC, +delta_weighted_premium NUMERIC, +is_smart_money BOOLEAN, + +-- DTE +dte INTEGER, +dte_bucket VARCHAR(20), +is_0dte BOOLEAN, + +-- Trade type +trade_type VARCHAR(20), -- 'SWEEP', 'BLOCK', 'CLUSTER', 'REGULAR' +is_sweep BOOLEAN, +is_block BOOLEAN, + +-- Index correlation +index_aligned BOOLEAN, +spy_flow_direction VARCHAR(10), +qqq_flow_direction VARCHAR(10), +vix_direction VARCHAR(10), + +-- Checklist +checklist_score INTEGER, +checklist_passed BOOLEAN, +checklist_details JSONB, + +-- Pattern tracking +pattern_hash VARCHAR(100), +historical_win_rate NUMERIC, +historical_avg_return NUMERIC, +pattern_confidence NUMERIC +``` + +--- + +## PART E — IMPLEMENTATION PRIORITY + +### Phase 1 (Highest Impact - Do First) +1. ✅ **Price Reaction Tracking** - Most important filter +2. ✅ **VWAP Integration** - Critical for entry/exit +3. ✅ **Signal Tier Classification** - Filter noise +4. ✅ **Trade Checklist** - Prevent bad trades + +### Phase 2 (High Value) +5. ✅ **Strike Clustering** - Identify institutional layering +6. ✅ **Delta Weighting** - Filter YOLO prints +7. ✅ **Index Correlation** - Context filter + +### Phase 3 (Nice to Have) +8. ✅ **Gamma Exposure** - Explains pinning behavior +9. ✅ **Sweep vs Block** - Trade type classification +10. ✅ **DTE Buckets** - Time-based filtering + +### Phase 4 (Analytics) +11. ✅ **Historical Win Rate** - Pattern analysis +12. ✅ **Enhanced Entry/Exit** - Refine trading logic + +--- + +## PART F — API ENDPOINT SUGGESTIONS + +### New Endpoints to Add + +1. **`GET /api/options-flow/enhanced`** + - Returns flow with all new enrichments + - Parameters: `include_price_reaction`, `include_gex`, etc. + +2. **`GET /api/options-flow/checklist`** + - Returns only signals that pass checklist (4/5 or 5/5) + +3. **`GET /api/options-flow/tier-1`** + - Returns only Tier-1 tradeable signals + +4. **`GET /api/patterns/stats`** + - Returns historical win rates per pattern + +5. **`GET /api/options-flow/vwap-analysis`** + - Returns VWAP-based entry opportunities + +--- + +## PART G — FRONTEND DISPLAY SUGGESTIONS + +### New UI Elements to Add + +1. **Signal Tier Badge** + - Display "TIER-1", "TIER-2", or "IGNORE" badge + - Color code: Green (Tier-1), Yellow (Tier-2), Gray (Ignore) + +2. **Price Reaction Indicator** + - Show 5m/15m price reaction percentage + - Green if positive reaction, Red if negative + - "Flow Led to Move" indicator + +3. **VWAP Distance Display** + - Show current price vs VWAP + - Visual indicator: Above/Below VWAP + - Entry opportunity: "VWAP Pullback" or "VWAP Reclaim" + +4. **Checklist Score Display** + - Show checklist score (X/5) + - Green if passed (4/5+), Red if failed + - Expandable details showing each check + +5. **Index Alignment Indicator** + - Show SPY/QQQ flow direction + - Show if aligned (green) or not (red) + +6. **Gamma Pin Level** + - Display GEX pin level on chart + - Show if price is near pin (resistance) + +7. **Strike Cluster Visualization** + - Show cluster size and total premium + - Highlight clustered strikes + +--- + +## PART H — TESTING SUGGESTIONS + +### Test Cases to Add + +1. **Price Reaction Tests** + - Test: Flow with no price reaction = should be filtered + - Test: Flow with 5m price reaction = should be tradeable + +2. **Tier Classification Tests** + - Test: 🟢 + 💎 + ⭐ + premium > 500K = Tier-1 + - Test: 🟢 + 💎 (no ⭐) = Tier-2 + - Test: OTM-only = Ignore + +3. **Checklist Tests** + - Test: 4/5 checks = passed + - Test: 3/5 checks = failed + +4. **VWAP Tests** + - Test: VWAP pullback entry detection + - Test: VWAP reclaim entry detection + +--- + +## SUMMARY + +### Key Takeaways + +1. **Price Reaction is #1 Priority** - This filters out hedges/rolls +2. **VWAP Integration is Critical** - Needed for proper entry/exit +3. **Tier Classification Reduces Noise** - Focus on Tier-1 signals +4. **Checklist Prevents Bad Trades** - Enforce 4/5 minimum +5. **Strike Clustering Identifies Institutions** - Multiple trades = stronger signal +6. **Index Correlation Adds Context** - Single stock works best with index alignment + +### Implementation Strategy + +- Start with Phase 1 (Price Reaction + VWAP + Tier Classification + Checklist) +- These 4 features will have the biggest impact on trade quality +- Then move to Phase 2 for additional filtering +- Phase 3 and 4 can be added over time as analytics improve + +### Expected Outcomes + +- **Higher Win Rate**: Filtering out hedges/rolls and low-quality signals +- **Better Entries**: VWAP-based entry logic +- **Better Exits**: Flow-based exit signals +- **Reduced Noise**: Tier classification and checklist +- **Institutional Detection**: Strike clustering and delta weighting + +--- + +**Note:** All suggestions are additive - they don't change existing code, just extend it with new services and enrichments. + diff --git a/backend/python_service/__pycache__/db.cpython-312.pyc b/backend/python_service/__pycache__/db.cpython-312.pyc index 544b702c987b8076eef3358e5144807b1db56910..797c742e9e74014fbc8b58e01a17799885efae3b 100644 GIT binary patch delta 1674 zcmY*ZTWnNC7@pa4*-Q7@?Pa(0a>};owxvdF6-zN<*f4+a_ z`~LZJc9-g(HG6;ccw7j^(r?}RMbC^kgm2EgRKLlV7UF`2H6bmINpT4yh{d!#W{=x3 zvLRx>&M(1$j7@bAiI`JDWiSj3acNRFtePAz)9lOq=9EoyoEPJ6&3PWhJ(`PnA>&Jl zS{bnZ)xAUl{64U5V9QV1qMn)i(_0?!`Yf3emZO;@#LtHtqH{18YDI*0LMfLJd!GCh zPfeQ-al`BloUqxP$XqAZyIG`^KiXufw%}x=j!xlgSgX(~wJNrul+~&)i`Vcc_yzM# z=(=qsdk%W>fq>9O;gax?Bq4Q;>Cpe;H_!!t1hM*(>d}`=QE3IkCPd%Fh_#j|E&Rgt z3#8SY!bdmqQiJQX+ENL?*OmCm`nCKO`8r-vI*VC*NfSKkPK#?i8rZ9)+$phi15}kO zLTPsvYxuPIp_s>cbV4|Q^7vaJkE81+Uh5-FF{qNHL}f%Vr^S#*$?I%XVfq*`a%}dh zxDWez84oHr@Q)!XnpkcQTK3g5QB2nZJ+Xef8E^*QU1b$D#Ryds)y4=QJF$3LBX=;GRHDTK4al7TE!(>dsRJ6G=&-Fq^P{zrA%xt zsXdoU8kv-SBu7=I8yQ8-XiA2#yg`pCkjaoF)8E(G$@Z^NG;FrWeSR)%IdWM-GwK+j zA@hvf&&&tPJBV0UcK$gt%eUg=4UQiEE)o>g@t zvqA_*2(z5A6G?KMS7HfSMx&DXy&Q2`m}DieRm?x-gRXZV;?2zdW%k(rohgVek=9@O$t8HH>17gvO!mWz|=Q{5b zy;y-j$`_gQMIIv&9LgK#%R7GW>??Hi7ZD!7o%5Yr zeu#fJ8NVstcHZgSSLhfj3=J1X4&EI}%#9=phf;;n@jD~5Fq|z4IFQ2&f)Ml;k>K$z zI*>p7LUrqnKx29*Vvh;9S0T*JZGV4$j*Zo7D0I0X*Wc#?`wKT!Au|Jgf?Sb%}&# zBohhB|KR8*4jtwhZmpW=)@q2gLZgTJQxrtwA;jH{Mv|tTO(<1q zOI!RwQR{$J6ucCQh7|ctC{xDxR?>MBu+3gut73X))9ALBp}oAoHv6&MMyeLVhWdhwai*< zhQM?iQWUJ=SidJF7|v-Dmkrl6ihB&V>4na^6fxv!(h^txC5Tzihe6r{&Aw4V^Uf?? z-)lRlnJ>^meeG5k1XG}uB>|y+IB^!`|JU!~iSze`&Bb>(A+Qb->NT3u=qjGWMx)_3 z0!HwOIEU}xDPVj>MCgGqD_)Z%lw3376zWE|Y ztBLG{GfbW#O=SbFBR)UmY*)cD>0mNz(jc36ZC8gb*dCpl$-J2u%H=W|VcqU+l0~4G z>F#rK7dZT1tuD6c{xh)RMZwU=fc7DveMeGIE-8&=5$tDwpyUshT0#{!L_d(E$~UAH z1$otpqvd*tep(GMWil3yb>f9~9NR+{c6ef)?3-u4bqiD~pon$?*s`9=% zuf5~AiCx52fLPNk{1+XzT;EPP0j+djs-??vC9Y zD~rh6vRbO&1iEt-B44QZNqLVQd@hZ$m%hM(2zPrp9OTdoV2d<2V)2h{ z7L?PduAuyn=spfrs;+1@XB0AK4?PY!3~Et;vWPKW7DP-cE`()vt0C;uu=>#X)ceR= PLEtW*VT%pB37_;Aan#MZ diff --git a/backend/python_service/__pycache__/main.cpython-312.pyc b/backend/python_service/__pycache__/main.cpython-312.pyc index a324234bc64d6460210b45f0d96bb4510c3e0b39..f6eb4d203d263111a114d89d0c0e4ac0e2e82a34 100644 GIT binary patch delta 1398 zcmZ`(OKclO7@pae*I8%VJnSYyQKpWAVj4`Nk_19YQ=3Ezlt&W;6h%TCdmT1%?R0h> z<0K{}g{TpgS~R6q2?yZ7L#2n-9JnDn5<)5m2NgxQk*I(Jw?-;LRpG!t>qLbEyP9vm z`JdnZJKE3FU(M>*>+9L#kzTuCZVQ7EhrB9lNPld zFr3Jmwi5|^X(MQ;N>)0Vp*~JMM*Kw9qO4a71gf%0X@=4VT}RHKX&B2F?5Nh(NBLX3 z@CXq5OMu_e-IHib@yZK$Q`OIED_W&_@P>M5)8EEk-PqNBTDrw9X`Ri#?Kq$LQ9ZKd zMZ3w}TWno>U{To4^sV?Rd!GG1UKty|p8W2WivOh>>dU;OzM@z5%x$Ri>+1YpTQZOT z+r9{`mvF2bza5Mc^eK*t(q}EezQIvhDIpPPN%TiGvDAX29;MVOao(nJdl!y+rBb*R zcv(2gi(t?X)-U60s zKbYB1b97KM@@SV1^cuM}dqK7qrs>nLfsib-{2W~t z(}}-c%Q?t~gCEQENgi-Jcs+j2nJn4{|2M31cs}jxRYK43_ltezOXTH3wLW>XA!x?BJ%4qNM{qD1mgroF_wvJ5|elGOwqCc=>!oV zBT#jUcI#Yc5xc}9_N&MoD7m?rx0aEsiX+(7$JIGR!EiDwzZ0(wP~sMAYEe;sktXY8 zPHttfA~m3pCi5+pg8clPTP(@>d3m=OQzsYji?O-^Wvn-M^LL1`-D1hiOUs`up{6P1 z05&2mwIn&C$O*_UvY#BRwvh7{dwF6}US?kU}HOlsUSAUs-k)zaf_`Wu_!UO7;H;X14y_LM7T~qWU9tG3&>>N{MK|4BcsFQ z3Fd8#8I#2=zR82bvj(ILdqH2`K02sHoz delta 23 dcmca#e!raWG%qg~0}uq7S!Mn)+sJp(8USGg2hjik diff --git a/backend/src/queries/optionsFlowQuery.js b/backend/src/queries/optionsFlowQuery.js index fb6ba88..61f0627 100644 --- a/backend/src/queries/optionsFlowQuery.js +++ b/backend/src/queries/optionsFlowQuery.js @@ -553,13 +553,14 @@ SELECT TO_CHAR(fe.flow_ts_local, 'HH12:MI:SS AM') AS "CreatedTime", /* Symbol line with session + catalyst + your badges */ + /* Use symbol_norm (clean symbol) instead of fe."Symbol" to avoid duplication */ ( CASE WHEN fe.direction = 'BULL' THEN '(' || fe.dir_count || '🟩) ' WHEN fe.direction = 'BEAR' THEN '(' || fe.dir_count || '🟥) ' ELSE '' END - || fe."Symbol" + || fe.symbol_norm || ' · ' || COALESCE(fe.session_bucket,'?') || CASE WHEN fe.near_alert_type IS NOT NULL THEN ' ⚡' ELSE '' END || CASE WHEN (fe.badge_round || fe.badge_more) <> '' THEN ' ' || fe.badge_round || fe.badge_more ELSE '' END diff --git a/backend/src/routes/optionsFlow.js b/backend/src/routes/optionsFlow.js index 977f229..a838df0 100644 --- a/backend/src/routes/optionsFlow.js +++ b/backend/src/routes/optionsFlow.js @@ -17,6 +17,7 @@ import { QueryProfiler } from '../utils/queryProfiler.js'; import { calculateFlowMomentum, getMomentumLabel } from '../services/momentumScore.js'; import { getMoneynessContext, isGammaSqueezeSetup, getMoneynessLabel } from '../utils/moneynessHelper.js'; import { batchFetchYahooFinanceData } from '../services/yahooFinanceService.js'; +import { fetchVolumeHistoryBulk } from '../services/volumeHistoryService.js'; const router = express.Router(); @@ -29,10 +30,16 @@ const USE_PYTHON_SERVICE = process.env.USE_PYTHON_SERVICE !== 'false'; router.get('/flow', async (req, res) => { try { - const { startDate, endDate, minPremium: minPremiumParam = 80000, tolPct = 0.20 } = req.query; + const { startDate, endDate, minPremium: minPremiumParam = 80000, tolPct = 0.20, skipPrices = 'false' } = req.query; // Parse minPremium to number (query params come as strings) const minPremium = parseFloat(minPremiumParam) || 80000; + // Ensure dates include full 24-hour day (00:00:00 to 23:59:59) + // SQL query uses date comparison, so we just need to ensure dates are properly formatted + // The SQL query's date BETWEEN clause will include the full day automatically + const normalizedStartDate = startDate ? startDate.split('T')[0] : null; // Extract just YYYY-MM-DD + const normalizedEndDate = endDate ? endDate.split('T')[0] : null; // Extract just YYYY-MM-DD + let rawData = []; // Try Python service first (if enabled) @@ -73,18 +80,58 @@ router.get('/flow', async (req, res) => { // Fallback to SQL if Python service not used or failed if (rawData.length === 0) { console.log('📊 Using SQL query (fallback or Python disabled)'); - console.log(`📅 Date range: ${startDate} to ${endDate}`); + const queryStartDate = normalizedStartDate || startDate; + const queryEndDate = normalizedEndDate || endDate; + console.log(`📅 Date range: ${queryStartDate} to ${queryEndDate} (full 24-hour days: 00:00:00 to 23:59:59)`); + + // First, check if there's any data in the database for these dates try { - // Note: SQL query has hardcoded minPremium=80000 in WHERE clause - // We'll filter by minPremium in JavaScript instead + const checkQuery = ` + SELECT COUNT(*) as total_count, + COUNT(CASE WHEN "Premium" IS NOT NULL AND TRIM("Premium"::text) <> '' THEN 1 END) as with_premium, + COUNT(CASE WHEN "StockEtf" = 'STOCK' THEN 1 END) as stocks, + MIN("CreatedDate") as min_date, + MAX("CreatedDate") as max_date + FROM public."OptionsFlow_monthly" + WHERE "CreatedDate"::date >= $1::date + AND "CreatedDate"::date <= $2::date + `; + const checkResult = await rawQuery(checkQuery, [normalizedStartDate || startDate, normalizedEndDate || endDate]); + if (checkResult && checkResult.length > 0) { + const stats = checkResult[0]; + console.log(`📊 Database stats for date range:`); + console.log(` Total rows: ${stats.total_count}`); + console.log(` With premium: ${stats.with_premium}`); + console.log(` Stocks: ${stats.stocks}`); + console.log(` Date range in DB: ${stats.min_date} to ${stats.max_date}`); + } + } catch (checkError) { + console.warn('⚠️ Could not check database stats:', checkError.message); + } + + try { + // Use SQL query as source of truth - it has restrictive filters // Profile the query execution + // Pass normalized dates to ensure full 24-hour day coverage (00:00:00 to 23:59:59) + const queryStartDate = normalizedStartDate || startDate; + const queryEndDate = normalizedEndDate || endDate; const { result, metrics } = await QueryProfiler.profile( - () => rawQuery(optionsFlowQuery, [startDate, endDate]), + () => rawQuery(optionsFlowQuery, [queryStartDate, queryEndDate]), 'optionsFlowQuery', { logSlowThreshold: 500 } // Log queries > 500ms ); rawData = result; - console.log(`✅ SQL query returned ${rawData.length} rows (${metrics.duration.toFixed(2)}ms)`); + console.log(`✅ SQL query (source of truth) returned ${rawData.length} rows (${metrics.duration.toFixed(2)}ms)`); + + if (rawData.length === 0) { + console.warn('⚠️ SQL query returned 0 rows. The SQL query (source of truth) has restrictive filters:'); + console.warn(' - premium_num > 80000'); + console.warn(' - badge_round IN (🟢,🔴)'); + console.warn(' - Must have 💎 (diamond) badge'); + console.warn(' - Must have ⭐ (star) badge'); + console.warn(' - Direction alignment (BULL/🟢/positive net OR BEAR/🔴/negative net)'); + console.warn(' These filters are defined in the SQL query and are not modified.'); + } // Add performance metrics to response metadata res.locals.queryMetrics = metrics; @@ -153,9 +200,9 @@ router.get('/flow', async (req, res) => { }); // Filter and sort - console.log(`📊 Raw data from SQL: ${rawData.length} rows`); + console.log(`📊 Raw data from SQL (source of truth): ${rawData.length} rows`); console.log(`📊 After enrichment: ${enrichedData.length} rows`); - console.log(`📊 Filtering with minPremium: ${minPremium}`); + console.log(`📊 SQL query already applied filters: premium > 80000, badges (🟢/🔴 + 💎 + ⭐), direction alignment`); // Debug: show sample of enriched data if (enrichedData.length > 0) { @@ -169,33 +216,25 @@ router.get('/flow', async (req, res) => { }); } - // Filter by premium (compare premium_num against minPremium) - const afterPremium = enrichedData.filter(row => { - const premium = parseFloat(row.premium_num) || 0; - return premium > minPremium; - }); - console.log(`📊 After premium filter (>${minPremium}): ${afterPremium.length} rows`); + // Note: SQL query is the source of truth and already applies all restrictive filters: + // - premium_num > 80000 + // - badge_round IN ('🟢','🔴') + // - Must have 💎 (diamond) badge + // - Must have ⭐ (star) badge + // - Direction alignment (BULL/🟢/positive net OR BEAR/🔴/negative net) + // So we don't need to re-apply these filters here - SQL query is authoritative - // Filter by badges - const afterBadges = afterPremium.filter(row => { - // Must have core badges - use badgesRaw object, not badges array - const badges = row.badgesRaw || {}; - const hasRound = badges.round === '🟢' || badges.round === '🔴'; - const hasDiamond = badges.more && badges.more.includes('💎'); - const hasStar = badges.more && badges.more.includes('⭐'); - return hasRound && hasDiamond && hasStar; - }); - console.log(`📊 After badge filter (🟢/🔴 + 💎 + ⭐): ${afterBadges.length} rows`); - - // Filter by direction/net premium alignment - const filtered = afterBadges.filter(row => { - // Direction must match net premium - const badges = row.badgesRaw || {}; - const netPrem = (row.bull_total || 0) - (row.bear_total || 0); - return (row.direction === 'BULL' && badges.round === '🟢' && netPrem > 0) || - (row.direction === 'BEAR' && badges.round === '🔴' && netPrem < 0); - }); - console.log(`📊 After direction/net premium filter: ${filtered.length} rows`); + // Only apply additional JavaScript-side filters if minPremium differs from SQL's hardcoded 80000 + let filtered = enrichedData; + if (minPremium !== 80000) { + filtered = filtered.filter(row => { + const premium = parseFloat(row.premium_num) || 0; + return premium > minPremium; + }); + console.log(`📊 Applied additional premium filter (>${minPremium}): ${filtered.length} rows`); + } else { + console.log(`📊 Using SQL query filters as-is (no additional filtering needed)`); + } // Sort by timestamp descending const sorted = filtered.sort((a, b) => { @@ -219,17 +258,52 @@ router.get('/flow', async (req, res) => { const flowReversals = batchDetectFlowReversals(tickerFlows, 30); const flowTrends = batchDetectFlowTrends(tickerFlows); - // Fetch stock price data for all unique symbols + // Fetch stock price data for all unique symbols (includes volume history) + // Skip if skipPrices=true for faster initial response const uniqueSymbols = [...new Set(sorted.map(row => (row.symbol_norm || row.Symbol || '').toUpperCase()).filter(Boolean))]; - console.log(`📈 Fetching stock prices for ${uniqueSymbols.length} symbols from Yahoo Finance...`); + const shouldSkipPrices = skipPrices === 'true' || skipPrices === true; let stockPrices = {}; + let volumeHistory = {}; + + if (!shouldSkipPrices) { + console.log(`📈 Fetching stock prices and volume history for ${uniqueSymbols.length} symbols from Yahoo Finance...`); try { stockPrices = await batchFetchYahooFinanceData(uniqueSymbols, 5); console.log(`✅ Successfully fetched stock prices for ${Object.keys(stockPrices).length} symbols`); + + // Extract volume history from stock price data + Object.keys(stockPrices).forEach(symbol => { + const priceData = stockPrices[symbol]; + if (priceData && priceData.volumeHistory && priceData.volumeHistory.length > 0) { + volumeHistory[symbol] = priceData.volumeHistory; + } + }); + console.log(`✅ Extracted volume history for ${Object.keys(volumeHistory).length} symbols`); } catch (error) { console.warn('⚠️ Error fetching stock prices from Yahoo Finance:', error.message); // Continue without stock prices if fetch fails + } + + // Fallback: Fetch volume history from database for symbols that didn't get data from Yahoo Finance + const missingSymbols = uniqueSymbols.filter(s => !volumeHistory[s]); + if (missingSymbols.length > 0) { + console.log(`📊 Fetching volume history from database for ${missingSymbols.length} symbols...`); + try { + const dbVolumeHistory = await fetchVolumeHistoryBulk(missingSymbols); + // Merge database results + Object.keys(dbVolumeHistory).forEach(symbol => { + if (!volumeHistory[symbol] && dbVolumeHistory[symbol].length > 0) { + volumeHistory[symbol] = dbVolumeHistory[symbol]; + } + }); + console.log(`✅ Fetched volume history from database for ${Object.keys(dbVolumeHistory).length} additional symbols`); + } catch (error) { + console.warn('⚠️ Error fetching volume history from database:', error.message); + } + } + } else { + console.log(`⏩ Skipping stock price/volume fetch for faster response (skipPrices=true)`); } // Add flow decay, reversal, and trend info to rows, then regenerate trade signals with trend data @@ -251,6 +325,9 @@ router.get('/flow', async (req, res) => { // Get stock price data for this symbol const stockPriceData = stockPrices[symbolUpper] || null; + // Get volume history for this symbol + const volumeHistoryData = volumeHistory[symbolUpper] || []; + const rowWithFlowInfo = { ...row, flowDecayRaw: decay, @@ -266,7 +343,9 @@ router.get('/flow', async (req, res) => { momentumColor: momentumLabel.color, momentumIcon: momentumLabel.icon, // Add stock price data - stockPrice: stockPriceData + stockPrice: stockPriceData, + // Add volume history (last 5 days) + volumeHistory: volumeHistoryData }; // Regenerate trade signal with flow trend data and moneyness context @@ -315,16 +394,21 @@ router.get('/flow', async (req, res) => { // Helper functions function formatSymbolDisplay(row, badges) { + // Use symbol_norm (raw symbol) instead of Symbol (which may already be formatted by SQL) + const rawSymbol = row.symbol_norm || (row.Symbol && !row.Symbol.includes('·') ? row.Symbol : null) || 'UNKNOWN'; + const dirCount = row.direction === 'BULL' ? `(${row.dir_count}🟩)` - : `(${row.dir_count}🟥)`; + : row.direction === 'BEAR' + ? `(${row.dir_count}🟥)` + : ''; const catalyst = row.near_alert_type ? ' ⚡' : ''; const badgeStr = (badges.round + badges.more + badges.flash).trim(); const fire = row.premium_num > 1000000 ? ' 🔥' : row.premium_num > 500000 ? ' 💵' : ''; - return `${dirCount} ${row.Symbol || row.symbol_norm} · ${row.session_bucket || '?'}${catalyst}${badgeStr ? ' ' + badgeStr : ''}${fire}`; + return `${dirCount}${dirCount ? ' ' : ''}${rawSymbol} · ${row.session_bucket || '?'}${catalyst}${badgeStr ? ' ' + badgeStr : ''}${fire}`; } function formatRocketDisplay(row, score, badges) { diff --git a/backend/src/routes/stockPrices.js b/backend/src/routes/stockPrices.js new file mode 100644 index 0000000..1065120 --- /dev/null +++ b/backend/src/routes/stockPrices.js @@ -0,0 +1,94 @@ +import express from 'express'; +import { batchFetchYahooFinanceData } from '../services/yahooFinanceService.js'; +import { fetchVolumeHistoryBulk } from '../services/volumeHistoryService.js'; + +const router = express.Router(); + +/** + * GET /api/stock-prices + * Fetch stock prices and volume history for multiple symbols + * Query params: symbols (comma-separated list) + */ +router.get('/', async (req, res) => { + try { + const { symbols } = req.query; + + if (!symbols) { + return res.status(400).json({ + success: false, + error: 'symbols parameter is required (comma-separated list)' + }); + } + + const symbolList = symbols.split(',').map(s => s.trim().toUpperCase()).filter(Boolean); + + if (symbolList.length === 0) { + return res.json({ + success: true, + data: {} + }); + } + + console.log(`📈 Fetching stock prices and volume for ${symbolList.length} symbols...`); + + let stockPrices = {}; + let volumeHistory = {}; + + // Fetch from Yahoo Finance + try { + stockPrices = await batchFetchYahooFinanceData(symbolList, 5); + console.log(`✅ Successfully fetched stock prices for ${Object.keys(stockPrices).length} symbols`); + + // Extract volume history from stock price data + Object.keys(stockPrices).forEach(symbol => { + const priceData = stockPrices[symbol]; + if (priceData && priceData.volumeHistory && priceData.volumeHistory.length > 0) { + volumeHistory[symbol] = priceData.volumeHistory; + } + }); + console.log(`✅ Extracted volume history for ${Object.keys(volumeHistory).length} symbols`); + } catch (error) { + console.warn('⚠️ Error fetching stock prices from Yahoo Finance:', error.message); + } + + // Fallback: Fetch volume history from database for symbols that didn't get data from Yahoo Finance + const missingSymbols = symbolList.filter(s => !volumeHistory[s]); + if (missingSymbols.length > 0) { + console.log(`📊 Fetching volume history from database for ${missingSymbols.length} symbols...`); + try { + const dbVolumeHistory = await fetchVolumeHistoryBulk(missingSymbols); + // Merge database results + Object.keys(dbVolumeHistory).forEach(symbol => { + if (!volumeHistory[symbol] && dbVolumeHistory[symbol].length > 0) { + volumeHistory[symbol] = dbVolumeHistory[symbol]; + } + }); + console.log(`✅ Fetched volume history from database for ${Object.keys(dbVolumeHistory).length} additional symbols`); + } catch (error) { + console.warn('⚠️ Error fetching volume history from database:', error.message); + } + } + + // Combine stock prices and volume history + const result = {}; + symbolList.forEach(symbol => { + result[symbol] = { + stockPrice: stockPrices[symbol] || null, + volumeHistory: volumeHistory[symbol] || [] + }; + }); + + res.json({ + success: true, + data: result, + timestamp: new Date().toISOString() + }); + + } catch (error) { + console.error('Stock prices error:', error); + res.status(500).json({ success: false, error: error.message }); + } +}); + +export default router; + diff --git a/backend/src/server.js b/backend/src/server.js index 039a049..db81c6d 100644 --- a/backend/src/server.js +++ b/backend/src/server.js @@ -4,6 +4,7 @@ import dotenv from 'dotenv'; import optionsFlowRouter from './routes/optionsFlow.js'; import dailyAnalysisRouter from './routes/dailyAnalysis.js'; import pricesRouter from './routes/prices.js'; +import stockPricesRouter from './routes/stockPrices.js'; import alertsRouter, { setupAlertsWebSocket } from './routes/alerts.js'; import scannerRouter from './routes/scanner.js'; import tradePlansRouter from './routes/tradePlans.js'; @@ -42,6 +43,7 @@ app.use(express.json()); app.use('/api/options', optionsFlowRouter); app.use('/api/analysis', dailyAnalysisRouter); app.use('/api/prices', pricesRouter); +app.use('/api/stock-prices', stockPricesRouter); app.use('/api/alerts', alertsRouter); app.use('/api/scanner', scannerRouter); app.use('/api/trade-plans', tradePlansRouter); diff --git a/backend/src/services/volumeHistoryService.js b/backend/src/services/volumeHistoryService.js new file mode 100644 index 0000000..4cbf879 --- /dev/null +++ b/backend/src/services/volumeHistoryService.js @@ -0,0 +1,144 @@ +/** + * Volume History Service + * Fetches last 5 days of volume data from database (used as fallback when Yahoo Finance data unavailable) + */ + +import { rawQuery } from '../db.js'; + +/** + * Fetch last 5 days of volume data for a single symbol + * @param {string} symbol - Stock symbol (e.g., 'AAPL') + * @returns {Promise} Array of volume data objects with date and volume + */ +export async function fetchVolumeHistory(symbol) { + try { + const query = ` + SELECT + "Date" as date, + volume, + close + FROM public.prices_daily + WHERE UPPER(symbol) = UPPER($1) + AND "Date" <= CURRENT_DATE + AND "Date" >= CURRENT_DATE - INTERVAL '5 days' + ORDER BY "Date" DESC + LIMIT 5 + `; + + const data = await rawQuery(query, [symbol]); + + // Format the data + return data.map(row => ({ + date: row.date, + volume: parseFloat(row.volume) || 0, + close: parseFloat(row.close) || 0 + })); + } catch (error) { + console.warn(`Failed to fetch volume history for ${symbol}:`, error.message); + return []; + } +} + +/** + * Batch fetch volume history for multiple symbols + * @param {string[]} symbols - Array of stock symbols + * @param {number} concurrency - Number of concurrent requests (default: 10) + * @returns {Promise} Map of symbol to volume history array + */ +export async function batchFetchVolumeHistory(symbols, concurrency = 10) { + if (!symbols || symbols.length === 0) { + return {}; + } + + // Remove duplicates and normalize symbols + const uniqueSymbols = [...new Set(symbols.map(s => s.toUpperCase().trim()))].filter(Boolean); + + if (uniqueSymbols.length === 0) { + return {}; + } + + const volumeHistoryMap = {}; + + // Process in batches to avoid overwhelming the database + for (let i = 0; i < uniqueSymbols.length; i += concurrency) { + const batch = uniqueSymbols.slice(i, i + concurrency); + const batchPromises = batch.map(async (symbol) => { + const data = await fetchVolumeHistory(symbol); + return { symbol, data }; + }); + + const batchResults = await Promise.allSettled(batchPromises); + + batchResults.forEach((result) => { + if (result.status === 'fulfilled') { + const { symbol, data } = result.value; + if (data && data.length > 0) { + volumeHistoryMap[symbol] = data; + } + } else { + console.warn(`Failed to fetch volume history for symbol in batch:`, result.reason); + } + }); + } + + return volumeHistoryMap; +} + +/** + * Fetch volume history for multiple symbols from database (fallback when Yahoo Finance unavailable) + * @param {string[]} symbols - Array of stock symbols + * @returns {Promise} Map of symbol to volume history array + */ +export async function fetchVolumeHistoryBulk(symbols) { + if (!symbols || symbols.length === 0) { + return {}; + } + + // Remove duplicates and normalize symbols + const uniqueSymbols = [...new Set(symbols.map(s => s.toUpperCase().trim()))].filter(Boolean); + + if (uniqueSymbols.length === 0) { + return {}; + } + + const volumeHistoryMap = {}; + + try { + const query = ` + SELECT + UPPER(symbol) as symbol, + "Date" as date, + volume, + close + FROM public.prices_daily + WHERE UPPER(symbol) = ANY($1::text[]) + AND "Date" <= CURRENT_DATE + AND "Date" >= CURRENT_DATE - INTERVAL '5 days' + ORDER BY symbol, "Date" DESC + `; + + const data = await rawQuery(query, [uniqueSymbols]); + + // Group by symbol + data.forEach(row => { + const symbol = row.symbol; + if (!volumeHistoryMap[symbol]) { + volumeHistoryMap[symbol] = []; + } + // Only keep the last 5 days per symbol + if (volumeHistoryMap[symbol].length < 5) { + volumeHistoryMap[symbol].push({ + date: row.date, + volume: parseFloat(row.volume) || 0, + close: parseFloat(row.close) || 0 + }); + } + }); + + return volumeHistoryMap; + } catch (error) { + console.warn('⚠️ Failed to fetch bulk volume history from database:', error.message); + return {}; + } +} + diff --git a/backend/src/services/yahooFinanceService.js b/backend/src/services/yahooFinanceService.js index 2c3b865..6585a8c 100644 --- a/backend/src/services/yahooFinanceService.js +++ b/backend/src/services/yahooFinanceService.js @@ -24,6 +24,37 @@ export async function fetchYahooFinanceData(symbol) { const result = data.chart.result[0]; const meta = result.meta; const quotes = result.indicators?.quote?.[0]; + const timestamps = result.timestamp || []; + + // Extract historical volume data (last 5 days) + const volumeHistory = []; + if (quotes && quotes.volume && timestamps.length > 0) { + const volumes = quotes.volume || []; + const closes = quotes.close || []; + const opens = quotes.open || []; + const highs = quotes.high || []; + const lows = quotes.low || []; + + // Get the last 5 days (or all available if less than 5) + const startIdx = Math.max(0, timestamps.length - 5); + for (let i = startIdx; i < timestamps.length; i++) { + if (volumes[i] !== null && volumes[i] !== undefined) { + const timestamp = timestamps[i]; + const date = new Date(timestamp * 1000); + volumeHistory.push({ + date: date.toISOString().split('T')[0], // Convert to YYYY-MM-DD + volume: Math.round(volumes[i]) || 0, + close: closes[i] || 0, + open: opens[i] || 0, + high: highs[i] || 0, + low: lows[i] || 0 + }); + } + } + + // Sort by date descending (most recent first) + volumeHistory.sort((a, b) => new Date(b.date) - new Date(a.date)); + } return { symbol: symbol.toUpperCase(), @@ -42,6 +73,8 @@ export async function fetchYahooFinanceData(symbol) { : 0, // Get recent price history recentPrices: quotes?.close?.slice(-5) || [], + // Get volume history (last 5 days) + volumeHistory: volumeHistory, timestamp: new Date().toISOString() }; } diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index a121a67..4593cd6 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -7,6 +7,7 @@ import PhaseClassifierPanel from '@/components/dashboard/PhaseClassifierPanel'; import AlertsFeed from '@/components/dashboard/AlertsFeed'; import Watchlist from '@/components/dashboard/Watchlist'; import PerformanceTrackingPanel from '@/components/dashboard/PerformanceTrackingPanel'; +import FlowInfoPanel from '@/components/dashboard/FlowInfoPanel'; export default function App() { return ( @@ -74,13 +75,18 @@ export default function App() { - {/* Right: Alerts Feed & Watchlist */} + {/* Right: Flow Info, Watchlist & Alerts Feed */}
- +
+ + {/* Bottom: Today's Signals */} +
+ +
); diff --git a/frontend/src/components/dashboard/FlowInfoPanel.jsx b/frontend/src/components/dashboard/FlowInfoPanel.jsx new file mode 100644 index 0000000..c6d5606 --- /dev/null +++ b/frontend/src/components/dashboard/FlowInfoPanel.jsx @@ -0,0 +1,186 @@ +import { useState } from 'react'; +import { Info, ChevronDown, ChevronUp } from 'lucide-react'; + +export default function FlowInfoPanel() { + const [isExpanded, setIsExpanded] = useState(false); + + return ( +
+ + + {isExpanded && ( +
+ {/* Badges Section */} +
+

🎯 Badges & Indicators

+
+
+ ⚡ +
+ Flash: + Aggressive sweep (AA/BB) with premium > $10K +
+
+
+ 🟢 +
+ Green Circle: + Bullish ITM premium dominance +
+
+
+ 🔴 +
+ Red Circle: + Bearish ITM premium dominance +
+
+
+ 💎 +
+ Diamond: + ITM premium dominance in direction +
+
+
+ ⭐ +
+ Star: + OTM flow spread > $10K +
+
+
+ 💰 +
+ Money: + Open Interest accumulation > $100K +
+
+
+ ✔ +
+ Check: + Volume > OI (new positioning) +
+
+
+ 🔥 +
+ Fire: + Premium > $1M +
+
+
+ 💵 +
+ Cash: + Premium > $500K +
+
+
+ 🚀 +
+ Rocket: + Single rocket - Vol > OI + (Flash OR Premium > $500K) +
+
+
+ 🚀🚀 +
+ Double Rocket: + Vol > OI + Flash + Money badge +
+
+
+ 🚀🚀🚀 +
+ Triple Rocket: + Vol > OI + Flash + Money + Premium > $500K +
+
+
+
+ + {/* Trading Terms */} +
+

📖 Trading Terms

+
+
+ Side: +
    +
  • • A/AA: Ask / Above Ask (aggressive buy)
    • +
  • • B/BB: Bid / Below Bid (aggressive sell)
    • +
+
+
+ Moneyness: +
    +
  • • ITM: In The Money
    • +
  • • OTM: Out The Money
    • +
+
+
+ Option Type: +
    +
  • • CALL: Right to buy at strike
    • +
  • • PUT: Right to sell at strike
    • +
+
+
+ Direction: +
    +
  • • BULL: Call Buy or Put Sell
    • +
  • • BEAR: Put Buy or Call Sell
    • +
+
+
+ Sessions: +
    +
  • • PRE: Pre-market (4:00 AM - 9:30 AM)
    • +
  • • RTH: Regular Trading Hours (9:30 AM - 4:00 PM)
    • +
  • • POST: After-hours (4:00 PM - 8:00 PM)
    • +
+
+
+ Key Metrics: +
    +
  • • Premium: Total dollar value of trade
    • +
  • • Volume: Contracts traded
    • +
  • • OI: Open Interest (outstanding contracts)
    • +
  • • Net Premium: Bull total - Bear total
    • +
+
+
+
+ + {/* Tape Alignment */} +
+

📊 Tape Alignment

+
+

+ ↗︎ (Up Arrow): Bullish flow with price moving up ≥ 0.20% +

+

+ ↘︎ (Down Arrow): Bearish flow with price moving down ≥ 0.20% +

+
+
+
+ )} +
+ ); +} + diff --git a/frontend/src/components/dashboard/OptionsFlowCardList.jsx b/frontend/src/components/dashboard/OptionsFlowCardList.jsx index 8680981..90ee76a 100644 --- a/frontend/src/components/dashboard/OptionsFlowCardList.jsx +++ b/frontend/src/components/dashboard/OptionsFlowCardList.jsx @@ -35,6 +35,7 @@ export function OptionsFlowCardList({ data, onCardClick, selectedRow }) { const momentum = row.momentumScore || 0; const signal = row.tradeSignal; const isSelected = selectedRow && (selectedRow.Symbol === row.Symbol || selectedRow.symbol_norm === row.symbol_norm); + const volumeHistory = row.volumeHistory || []; // Date fields const createdDate = row.CreatedDate; @@ -390,6 +391,74 @@ export function OptionsFlowCardList({ data, onCardClick, selectedRow }) { + + {/* Last 5 Days Volume History - Table Format (New Row) */} + {volumeHistory && volumeHistory.length > 0 && ( +
+
+ Last 5 Days Volume +
+ + + + + + + + + + + + + {volumeHistory.map((day, idx) => { + const date = new Date(day.date); + const today = new Date(); + today.setHours(0, 0, 0, 0); + const yesterday = new Date(today); + yesterday.setDate(yesterday.getDate() - 1); + const dayDate = new Date(date); + dayDate.setHours(0, 0, 0, 0); + + let dayLabel; + if (dayDate.getTime() === today.getTime()) { + dayLabel = 'Today'; + } else if (dayDate.getTime() === yesterday.getTime()) { + dayLabel = 'Yesterday'; + } else { + dayLabel = date.toLocaleDateString('en-US', { month: 'short', day: 'numeric' }); + } + + // Format volume with K/M/B + const formatVolume = (vol) => { + if (!vol || vol === 0) return '—'; + const absVal = Math.abs(vol); + if (absVal >= 1e9) return `${(absVal / 1e9).toFixed(2)}B`; + if (absVal >= 1e6) return `${(absVal / 1e6).toFixed(2)}M`; + if (absVal >= 1e3) return `${(absVal / 1e3).toFixed(2)}K`; + return absVal.toLocaleString(); + }; + + // Format price + const formatPrice = (price) => { + if (!price || price === 0) return '—'; + return `$${parseFloat(price).toFixed(2)}`; + }; + + return ( + + + + + + + + + ); + })} + +
DateVolumeOpenHighLowClose
{dayLabel}{formatVolume(day.volume)}{formatPrice(day.open)}{formatPrice(day.high)}{formatPrice(day.low)}{formatPrice(day.close)}
+
+ )} ); })} diff --git a/frontend/src/components/dashboard/OptionsFlowPanel.jsx b/frontend/src/components/dashboard/OptionsFlowPanel.jsx index c9f4ecd..926363d 100644 --- a/frontend/src/components/dashboard/OptionsFlowPanel.jsx +++ b/frontend/src/components/dashboard/OptionsFlowPanel.jsx @@ -1,5 +1,6 @@ import { useEffect, useState, useMemo, useRef } from 'react'; import { useOptionsFlow } from '@/hooks/useOptionsFlow'; +import { useStockPrices } from '@/hooks/useStockPrices'; import { DataTable } from '@/components/ui/data-table'; import { Badge } from '@/components/ui/Badge'; import { Input } from '@/components/ui/input'; @@ -73,22 +74,144 @@ export default function OptionsFlowPanel() { return score; }; - // Rank data by best trade score + // Extract unique symbols for stock price fetching + const uniqueSymbols = useMemo(() => { + if (!data) return []; + return [...new Set(data.map(row => (row.symbol_norm || row.Symbol || '').toUpperCase()).filter(Boolean))]; + }, [data]); + + // Fetch stock prices and volume asynchronously + const { data: stockPricesData } = useStockPrices(uniqueSymbols); + + // Rank data by best trade score and merge stock price/volume data const filteredData = useMemo(() => { if (!data) return []; // Add best trade score and sort by it - const filtered = data.map(row => ({ + const filtered = data.map(row => { + const symbolUpper = (row.symbol_norm || row.Symbol || '').toUpperCase(); + const stockPriceInfo = stockPricesData[symbolUpper]; + + // Merge stock price and volume history data + const mergedRow = { ...row, bestTradeScore: calculateBestTradeScore(row) - })).sort((a, b) => b.bestTradeScore - a.bestTradeScore); + }; + + // Add stock price data if available + if (stockPriceInfo) { + if (stockPriceInfo.stockPrice) { + mergedRow.stockPrice = stockPriceInfo.stockPrice; + } + if (stockPriceInfo.volumeHistory && stockPriceInfo.volumeHistory.length > 0) { + mergedRow.volumeHistory = stockPriceInfo.volumeHistory; + } + } + + return mergedRow; + }).sort((a, b) => b.bestTradeScore - a.bestTradeScore); return filtered; - }, [data]); + }, [data, stockPricesData]); // Get top 5 trades for summary + // Filter: items repeated more than once, more than 2 rockets, highest premium const topTrades = useMemo(() => { - return filteredData.slice(0, 5); + if (!filteredData || filteredData.length === 0) return []; + + // Helper function to extract clean symbol + const getCleanSymbol = (row) => { + // Prioritize symbol_norm (clean, normalized) + if (row.symbol_norm) { + return row.symbol_norm.toUpperCase().trim(); + } + // If Symbol is formatted like "(35) (35) HOOD · RTH ⚡ 🟢💎⭐💰⚡ 🔥", extract just the symbol + if (row.Symbol) { + const symbolStr = String(row.Symbol); + // Try to extract symbol from formatted string (look for pattern like "HOOD ·" or just "HOOD") + // The symbol is usually before the "·" separator + const match = symbolStr.match(/([A-Z]{1,5})\s*·/); + if (match && match[1]) { + return match[1].toUpperCase().trim(); + } + // If no separator, try to find a stock symbol (1-5 uppercase letters) + const symbolMatch = symbolStr.match(/\b([A-Z]{1,5})\b/); + if (symbolMatch && symbolMatch[1]) { + return symbolMatch[1].toUpperCase().trim(); + } + // Fallback: use the whole string if it's short and looks like a symbol + const cleaned = symbolStr.replace(/[^A-Z]/g, '').toUpperCase(); + if (cleaned.length >= 1 && cleaned.length <= 5) { + return cleaned; + } + } + return ''; + }; + + // Count symbol occurrences using clean symbols + const symbolCounts = {}; + filteredData.forEach(row => { + const symbol = getCleanSymbol(row); + if (symbol) { + symbolCounts[symbol] = (symbolCounts[symbol] || 0) + 1; + } + }); + + // Helper function to count rockets + const countRockets = (rocketStr) => { + if (!rocketStr) return 0; + // Count occurrences of 🚀 emoji + return (rocketStr.match(/🚀/g) || []).length; + }; + + // Helper function to get premium as number + const getPremiumNum = (row) => { + // Try premium_num first (raw numeric value) + if (row.premium_num !== undefined && row.premium_num !== null) { + return parseFloat(row.premium_num) || 0; + } + // Try Premium field (might be formatted string like "500 K" or "1.2 M") + if (row.Premium) { + const premiumStr = String(row.Premium).trim(); + // Try to parse if it's a number string + const num = parseFloat(premiumStr.replace(/[^0-9.-]/g, '')); + if (!isNaN(num)) { + // Check for K or M suffix + if (premiumStr.toUpperCase().includes('M')) { + return num * 1000000; + } else if (premiumStr.toUpperCase().includes('K')) { + return num * 1000; + } + return num; + } + } + return 0; + }; + + // Filter: items that appear more than once AND have 2 or more rockets (>= 2, so 🚀🚀 or 🚀🚀🚀) + const filtered = filteredData.filter(row => { + const symbol = getCleanSymbol(row); + // Check multiple rocket field variations + const rocket = row.Rocket || row.rocketDisplay || row.rocket || row.rocket_with_mny || ''; + const rocketCount = countRockets(rocket); + + // Must appear more than once (duplicate) + const isRepeated = symbolCounts[symbol] > 1; + // Must have 2 or more rockets (count >= 2, so 🚀🚀 or 🚀🚀🚀) + const has2OrMoreRockets = rocketCount >= 2; + + return isRepeated && has2OrMoreRockets; + }); + + // Sort by premium (highest first) + filtered.sort((a, b) => { + const premiumA = getPremiumNum(a); + const premiumB = getPremiumNum(b); + return premiumB - premiumA; + }); + + // Return top 5 + return filtered.slice(0, 5); }, [filteredData]); const handleRowClick = (row) => { @@ -776,16 +899,22 @@ export default function OptionsFlowPanel() { {/* Top Trades Summary */} - {!loading && !error && topTrades.length > 0 && ( + {!loading && !error && (

🏆 TOP 5 TRADES FOR TODAY

- Ranked by best trade potential + + {topTrades.length > 0 + ? `Filtered: Repeated symbols with 2+ rockets, sorted by highest premium` + : `No trades found matching criteria (repeated symbols with 2+ rockets)`} +
-
- {topTrades.map((row, idx) => { + {topTrades.length > 0 ? ( +
+ {topTrades.map((row, idx) => { const signal = row.tradeSignal; - const symbol = row.Symbol || row.symbol_norm; + // Use clean symbol extraction (same logic as filtering) + const symbol = row.symbol_norm || (row.Symbol && row.Symbol.match(/([A-Z]{1,5})\s*·/)?.[1]) || row.Symbol || 'UNKNOWN'; const score = row.bestTradeScore || 0; const hasSignal = signal && signal.signal !== 'NEUTRAL' && signal.signal !== 'WAIT'; @@ -828,7 +957,15 @@ export default function OptionsFlowPanel() { ); })} -
+
+ ) : ( +
+

No trades found that meet the criteria:

+

• Symbol appears more than once in the data

+

• Has 2 or more rockets (🚀🚀 or 🚀🚀🚀)

+

• Sorted by highest premium

+
+ )}
)} diff --git a/frontend/src/components/dashboard/PerformanceTrackingPanel.jsx b/frontend/src/components/dashboard/PerformanceTrackingPanel.jsx index c10f485..eb88aaf 100644 --- a/frontend/src/components/dashboard/PerformanceTrackingPanel.jsx +++ b/frontend/src/components/dashboard/PerformanceTrackingPanel.jsx @@ -39,33 +39,40 @@ export default function PerformanceTrackingPanel() { const result = await response.json(); + // Check if response has the expected trading stats format if (result.success && result.data) { setStats({ totalSignals: result.data.totalSignals || 0, highConviction: result.data.highConviction || 0, currentlyTracking: result.data.currentlyTracking || 0, winRate: { - all: result.data.winRate?.all || 0, - highScore: result.data.winRate?.highScore || 0, - tapeAligned: result.data.winRate?.tapeAligned || 0, - patternMatched: result.data.winRate?.patternMatched || 0 + all: result.data.winRate?.all || stats.winRate.all, + highScore: result.data.winRate?.highScore || stats.winRate.highScore, + tapeAligned: result.data.winRate?.tapeAligned || stats.winRate.tapeAligned, + patternMatched: result.data.winRate?.patternMatched || stats.winRate.patternMatched }, avgPerformance: { - avgWinner: result.data.avgPerformance?.avgWinner || 0, - avgLoser: result.data.avgPerformance?.avgLoser || 0, - rrRatio: result.data.avgPerformance?.rrRatio || 0, - expectancy: result.data.avgPerformance?.expectancy || 0 + avgWinner: result.data.avgPerformance?.avgWinner || stats.avgPerformance.avgWinner, + avgLoser: result.data.avgPerformance?.avgLoser || stats.avgPerformance.avgLoser, + rrRatio: result.data.avgPerformance?.rrRatio || stats.avgPerformance.rrRatio, + expectancy: result.data.avgPerformance?.expectancy || stats.avgPerformance.expectancy } }); + } else if (result.success && result.stats) { + // Endpoint exists but returns query performance stats, not trading stats + // Keep default values - this endpoint is for query metrics, not trading stats + console.warn('Performance endpoint returned query stats, not trading stats. Using default values.'); } else { - throw new Error(result.error || 'Failed to fetch stats'); + // If endpoint doesn't return expected format, silently use defaults + console.warn('Performance stats endpoint returned unexpected format. Using default values.'); } } catch (error) { - // Only log if it's not a 404 (endpoint doesn't exist) - if (!error.message.includes('404')) { - console.error('Failed to fetch performance stats:', error); + // Silently handle errors - endpoint may not exist or may be unavailable + // Keep existing default stats on error (don't reset to 0) + // Only log non-network errors for debugging + if (error.name !== 'TypeError' && !error.message.includes('fetch')) { + console.warn('Performance stats endpoint error:', error.message); } - // Keep existing stats on error (don't reset to 0) } }; diff --git a/frontend/src/hooks/useOptionsFlow.js b/frontend/src/hooks/useOptionsFlow.js index fa35615..c36529f 100644 --- a/frontend/src/hooks/useOptionsFlow.js +++ b/frontend/src/hooks/useOptionsFlow.js @@ -33,6 +33,9 @@ export function useOptionsFlow({ autoRefresh = false, interval = 30000, ...filte ) }); + // Add skipPrices=true for faster initial response + params.append('skipPrices', 'true'); + const response = await fetch(`${getApiUrl('/api/options/flow')}?${params}`); const result = await response.json(); diff --git a/frontend/src/hooks/useStockPrices.js b/frontend/src/hooks/useStockPrices.js new file mode 100644 index 0000000..b17c1b6 --- /dev/null +++ b/frontend/src/hooks/useStockPrices.js @@ -0,0 +1,61 @@ +import { useState, useEffect, useCallback } from 'react'; +import { getApiUrl } from '@/config/api'; + +/** + * Hook to fetch stock prices and volume history for symbols + * @param {string[]} symbols - Array of stock symbols to fetch + * @returns {Object} { data, loading, error, refetch } + */ +export function useStockPrices(symbols) { + const [data, setData] = useState({}); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(null); + + const fetchStockPrices = useCallback(async (symbolList) => { + if (!symbolList || symbolList.length === 0) { + setData({}); + return; + } + + try { + setLoading(true); + setError(null); + + // Remove duplicates and normalize + const uniqueSymbols = [...new Set(symbolList.map(s => s.toUpperCase().trim()))].filter(Boolean); + + if (uniqueSymbols.length === 0) { + setData({}); + setLoading(false); + return; + } + + const params = new URLSearchParams({ + symbols: uniqueSymbols.join(',') + }); + + const response = await fetch(`${getApiUrl('/api/stock-prices')}?${params}`); + const result = await response.json(); + + if (result.success) { + setData(result.data || {}); + } else { + throw new Error(result.error || 'Failed to fetch stock prices'); + } + } catch (err) { + console.error('Error fetching stock prices:', err); + setError(err.message); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + if (symbols && symbols.length > 0) { + fetchStockPrices(symbols); + } + }, [symbols, fetchStockPrices]); + + return { data, loading, error, refetch: () => fetchStockPrices(symbols) }; +} + diff --git a/watch_and_upload.py b/watch_and_upload.py new file mode 100644 index 0000000..e69de29 diff --git a/watch_and_upload_blackbox_postgres.py b/watch_and_upload_blackbox_postgres.py new file mode 100644 index 0000000..4058dc5 --- /dev/null +++ b/watch_and_upload_blackbox_postgres.py @@ -0,0 +1,733 @@ +# sync_blackbox_flow.py — BlackBox API sync to PostgreSQL +# Fetches options flow data from BlackBox API and syncs to PostgreSQL database + +import os, time, hashlib, re, argparse +from pathlib import Path +from typing import Dict, Optional, List +from datetime import datetime, date +import pandas as pd +import numpy as np +import requests +import json +import psycopg2 +from psycopg2.extras import execute_values +from psycopg2 import sql +from dotenv import load_dotenv + +load_dotenv() + +# ───────────────────────────────────────────── +# Config +# ───────────────────────────────────────────── +WRITE_POSTGRES = True + +# BlackBox API Configuration +BLACKBOX_API_URL = "https://api.blackboxstocks.com/api/v2/options/getFlowMobile" +BLACKBOX_API_TOKEN = os.getenv("BLACKBOX_API_TOKEN", "eyJhbGciOiJodHRwOi8vd3d3LnczLm9yZy8yMDAxLzA0L3htbGRzaWctbW9yZSNobWFjLXNoYTI1NiIsInR5cCI6IkpXVCJ9.eyJodHRwOi8vc2NoZW1hcy54bWxzb2FwLm9yZy93cy8yMDA1LzA1L2lkZW50aXR5L2NsYWltcy9uYW1lIjoic3JlZGR5MDgxOUBnbWFpbC5jb20iLCJodHRwOi8vc2NoZW1hcy54bWxzb2FwLm9yZy93cy8yMDA1LzA1L2lkZW50aXR5L2NsYWltcy9uYW1laWRlbnRpZmllciI6Ijk3ZDA5NThjLTBhOTktNDA5OC05OWQwLWNkYTg3Njg5N2Q3ZiIsImh0dHA6Ly9zY2hlbWFzLnhtbHNvYXAub3JnL3dzLzIwMDUvMDUvaWRlbnRpdHkvY2xhaW1zL3NpZCI6InBpS1plMktZN2d2NnErbDZyM3g3YkpHYU9ESlUrZVdOZmMwc1NicWZZNU09IiwiaHR0cDovL3NjaGVtYXMubWljcm9zb2Z0LmNvbS93cy8yMDA4LzA2L2lkZW50aXR5L2NsYWltcy9yb2xlIjpbIlN1YnNjcmliZXIiLCJPcHRpb25zIiwiRXF1aXRpZXMiXSwiZXhwIjoxNzY1NzQ4NjI1fQ.Evq__DugD7s1kAytUqtsknQFIeOPjKM3iwp6cDI0hJI") + +# PostgreSQL connection parameters +POSTGRES_HOST = os.getenv("POSTGRES_HOST", "localhost") +POSTGRES_PORT = int(os.getenv("POSTGRES_PORT", "5432")) +POSTGRES_DB = os.getenv("POSTGRES_DB", "institutional_trader") +POSTGRES_USER = os.getenv("POSTGRES_USER", "postgres") +POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "postgres") +POSTGRES_SCHEMA = os.getenv("POSTGRES_SCHEMA", "public") + +# ⚠️ Per-table schema style: 'snake' or 'camel' +TABLE_STYLE = { + "AlertStream": "snake", + "AlertStream_monthly": "snake", + "OptionsFlow": "camel", # ← your Supabase columns are CamelCase here + "OptionsFlow_monthly": "camel", # ← same + "OptionsVolume": "camel", # adjust if needed + "Short_Long": "camel", # adjust if needed +} + +_TARGET_STEMS = set(TABLE_STYLE.keys()) + +# Upsert behavior / logging +INCREMENTAL = True +SINGLE_CHUNK = False # chunked to avoid 57014 +UPSERT_CHUNK = 3000 # drop to 1000 if still timeouts +MAX_RETRIES = 3 +PRINT_PROGRESS = False +PRINT_PRUNE_LOGS = False +USE_RETURNING_MINIMAL= True + +print_flush = lambda *a, **k: print(*a, **k, flush=True) +def _ts(): return time.strftime("%Y-%m-%d %H:%M:%S") + +# ───────────────────────────────────────────── +# PostgreSQL connection +# ───────────────────────────────────────────── +def _pg_conn(): + """Create a PostgreSQL connection.""" + return psycopg2.connect( + host=POSTGRES_HOST, + port=POSTGRES_PORT, + database=POSTGRES_DB, + user=POSTGRES_USER, + password=POSTGRES_PASSWORD, + options=f"-c search_path={POSTGRES_SCHEMA}" + ) + +# ───────────────────────────────────────────── +# Expected headers (both styles) + mappings +# ───────────────────────────────────────────── +# OptionsFlow (CamelCase) +EXPECTED_OPTIONSFLOW_CAMEL = [ + "CreatedDate","CreatedTime","Symbol","Type","Volume","Price","Side", + "CallPut","Strike","Spot","Premium","ExpirationDate","Color", + "ImpliedVolatility","Dte","ER","StockEtf","Sector","Uoa", + "Weekly","MktCap","OI" +] +# OptionsFlow (snake_case) +EXPECTED_OPTIONSFLOW_SNAKE = [ + "created_date","created_time","symbol","type","volume","price","side", + "callput","strike","spot","premium","expiration_date","color", + "implied_volatility","dte","er","stock_etf","sector","uoa", + "weekly","mktcap","oi" +] +# AlertStream (CamelCase in files → snake in DB) +EXPECTED_ALERTSTREAM_SNAKE = [ + "date","timestamp","ticker","volume","price","pct_of_avg30", + "notional","message","type","securitytype","industry","sector", + "avg30day","float","earningsdate" +] + +# Camel→snake renames (if file arrives CamelCase but DB expects snake) +MAP_OPTIONSFLOW_SNAKE = { + "CreatedDate":"created_date","CreatedTime":"created_time","Symbol":"symbol","Type":"type", + "Volume":"volume","Price":"price","Side":"side","CallPut":"callput","Strike":"strike", + "Spot":"spot","Premium":"premium","ExpirationDate":"expiration_date","Color":"color", + "ImpliedVolatility":"implied_volatility","Dte":"dte","ER":"er","StockEtf":"stock_etf", + "Sector":"sector","Uoa":"uoa","Weekly":"weekly","MktCap":"mktcap","OI":"oi" +} +MAP_ALERTSTREAM_SNAKE = { + "Date":"date","Timestamp":"timestamp","Ticker":"ticker","Volume":"volume","Price":"price", + "Pct_of_Avg30Day":"pct_of_avg30","Notional":"notional","Message":"message","Type":"type", + "SecurityType":"securitytype","Industry":"industry","Sector":"sector", + "Avg30Day":"avg30day","Float":"float","EarningsDate":"earningsdate" +} + +# snake→Camel renames (if file is snake but DB expects Camel) +MAP_OPTIONSFLOW_CAMEL = {v:k for k,v in MAP_OPTIONSFLOW_SNAKE.items()} + +# ───────────────────────────────────────────── +# BlackBox API Integration +# ───────────────────────────────────────────── +def build_filter_bitmask(_filter_options: Optional[Dict] = None) -> int: + """Build the filter bitmask based on filter options.""" + # Default filter value that enables common filters + return 2198487171967 + +def build_filters(start_date: datetime, end_date: datetime, custom_filters: Optional[Dict] = None) -> Dict: + """Build default filters object.""" + date_str = start_date.isoformat() + + filters = { + "optionsDate": { + "start": date_str, + "end": end_date.isoformat() + }, + "expireOptionsDate": { + "start": date_str, + "end": end_date.isoformat() + }, + "optionsFlowPuts": True, + "optionsFlowCalls": True, + "optionsFlowYellow": True, + "optionsFlowWhite": True, + "optionsFlowMagenta": True, + "optionsFlowAboveAskOnly": True, + "optionsFlowBelowBidOnly": True, + "optionsFlowAtOrAboveAsk": True, + "optionsFlowAtOrBelowBid": True, + "optionsFlowMultileg": False, + "optionsFlowOnlyMultiLeg": False, + "optionsFlowBelowPoint5": False, + "optionsFlowBelow5": False, + "optionsFlow100Contracts": False, + "optionsFlow500Contracts": False, + "optionsFlow5000Contracts": False, + "optionsFlowStock": True, + "optionsFlowEtf": True, + "optionsFlowAbove50k": False, + "optionsFlowAbove100k": False, + "optionsFlowAbove200k": False, + "optionsFlowAbove500k": False, + "optionsFlowAbove1m": False, + "marketCapAbove750B": False, + "optionsFlowInTheMoney": False, + "optionsFlowOutOfTheMoney": False, + "optionsFlowSweepOnly": False, + "optionsFlowWeeklyOnly": False, + "optionsFlowEarningsReportOnly": False, + "optionsFlowUnusualOnly": False, + "optionsFlowExDiv": False, + "optionsFlowConsumerDiscretionary": True, + "optionsFlowIndustrials": True, + "optionsFlowInformationTechnology": True, + "optionsFlowRealEstate": True, + "optionsFlowHealthCare": True, + "optionsFlowEnergy": True, + "optionsFlowFinancials": True, + "optionsFlowMaterials": True, + "optionsFlowConsumerStaples": True, + "optionsFlowCommunicationServices": True, + "optionsFlowUtilities": True, + "optionsExpirationRange": False, + "optionsFlowSectorNone": True, + } + + if custom_filters: + filters.update(custom_filters) + + return filters + +# Constants +TIMEZONE_SUFFIX = "+00:00" + +def fetch_blackbox_flow(options: Optional[Dict] = None) -> List[Dict]: + """Fetch options flow data from BlackBox Stocks API.""" + if options is None: + options = {} + + if not BLACKBOX_API_TOKEN: + raise ValueError( + "BLACKBOX_API_TOKEN not found in environment variables.\n" + "Please add BLACKBOX_API_TOKEN to your .env file or set it as an environment variable." + ) + + # Parse dates - default to today if not provided + if options.get("startDate"): + if isinstance(options["startDate"], str): + start_date = datetime.fromisoformat(options["startDate"].replace("Z", TIMEZONE_SUFFIX)) + elif isinstance(options["startDate"], date): + start_date = datetime.combine(options["startDate"], datetime.min.time()) + else: + start_date = options["startDate"] + else: + start_date = datetime.now() + + if options.get("endDate"): + if isinstance(options["endDate"], str): + end_date = datetime.fromisoformat(options["endDate"].replace("Z", TIMEZONE_SUFFIX)) + elif isinstance(options["endDate"], date): + end_date = datetime.combine(options["endDate"], datetime.max.time()) + else: + end_date = options["endDate"] + else: + end_date = start_date + + # Build request body + body = { + "historical": options.get("historical", False), + "symbol": options.get("symbol", ""), + "strike": options.get("strike", 0), + "count": options.get("count") or options.get("limit", 300), + "filter": build_filter_bitmask(options.get("filters")), + "filters": build_filters(start_date, end_date, options.get("filters")), + "fromDate": start_date.isoformat(), + "toDate": end_date.isoformat() + } + + try: + response = requests.post( + BLACKBOX_API_URL, + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"Bearer {BLACKBOX_API_TOKEN}" + }, + json=body, + timeout=30 + ) + + if not response.ok: + error_text = response.text + raise RuntimeError( + f"BlackBox API error: {response.status_code} {response.reason}\n" + f"Response: {error_text}" + ) + + data = response.json() + + # Handle different response structures + if isinstance(data, list): + return data + elif isinstance(data, dict): + if "data" in data and isinstance(data["data"], list): + return data["data"] + elif "flows" in data and isinstance(data["flows"], list): + return data["flows"] + elif "results" in data and isinstance(data["results"], list): + return data["results"] + else: + print_flush(f"{_ts()} | ⚠️ Unexpected API response structure: {list(data.keys())}") + return [] + else: + return [] + except Exception as e: + print_flush(f"{_ts()} | ❌ Error fetching BlackBox flow data: {e}") + raise + +def map_blackbox_to_database(api_record: Dict) -> Dict: + """Map BlackBox API response to database schema.""" + def get_value(obj: Dict, *keys: str) -> Optional[str]: + """Safely extract values from object.""" + for key in keys: + if key in obj and obj[key] is not None: + return str(obj[key]) + return None + + def format_date(date_str: Optional[str]) -> Optional[str]: + """Format date as YYYY-MM-DD.""" + if not date_str: + return None + try: + dt = datetime.fromisoformat(str(date_str).replace("Z", TIMEZONE_SUFFIX)) + return dt.strftime("%Y-%m-%d") + except (ValueError, AttributeError): + try: + dt = datetime.strptime(str(date_str), "%Y-%m-%d") + return dt.strftime("%Y-%m-%d") + except (ValueError, AttributeError): + return str(date_str) + + def format_time(time_str: Optional[str]) -> Optional[str]: + """Format time string.""" + if not time_str: + return None + return str(time_str) + + # Map fields - try multiple possible field names from API + mapped = { + "CreatedDate": format_date( + get_value(api_record, "createdDate", "CreatedDate", "date", "Date", "timestamp", "Timestamp") + ), + "CreatedTime": format_time( + get_value(api_record, "createdTime", "CreatedTime", "time", "Time", "timestamp", "Timestamp") + ), + "Symbol": get_value(api_record, "symbol", "Symbol", "ticker", "Ticker", "underlying", "Underlying"), + "Type": get_value(api_record, "type", "Type", "tradeType", "TradeType"), + "Volume": get_value(api_record, "volume", "Volume", "vol", "Vol", "contracts", "Contracts"), + "Price": get_value(api_record, "price", "Price", "lastPrice", "LastPrice", "tradePrice", "TradePrice"), + "Side": get_value(api_record, "side", "Side", "tradeSide", "TradeSide", "direction", "Direction"), + "CallPut": get_value(api_record, "callPut", "CallPut", "optionType", "OptionType", "putCall", "PutCall", "type", "Type"), + "Strike": get_value(api_record, "strike", "Strike", "strikePrice", "StrikePrice"), + "Spot": get_value(api_record, "spot", "Spot", "underlyingPrice", "UnderlyingPrice", "stockPrice", "StockPrice"), + "Premium": get_value(api_record, "premium", "Premium", "totalPremium", "TotalPremium", "notional", "Notional"), + "ExpirationDate": format_date( + get_value(api_record, "expirationDate", "ExpirationDate", "expiry", "Expiry", "expiration", "Expiration") + ), + "Color": get_value(api_record, "color", "Color", "tradeColor", "TradeColor"), + "ImpliedVolatility": get_value(api_record, "impliedVolatility", "ImpliedVolatility", "iv", "IV", "volatility", "Volatility"), + "Dte": get_value(api_record, "dte", "Dte", "DTE", "daysToExpiration", "DaysToExpiration", "daysToExpiry", "DaysToExpiry"), + "ER": get_value(api_record, "er", "ER", "earnings", "Earnings", "earningsReport", "EarningsReport"), + "StockEtf": get_value(api_record, "stockEtf", "StockEtf", "assetType", "AssetType", "securityType", "SecurityType"), + "Sector": get_value(api_record, "sector", "Sector", "industry", "Industry"), + "Uoa": get_value(api_record, "uoa", "Uoa", "UOA", "underlyingOfAsset", "UnderlyingOfAsset"), + "Weekly": get_value(api_record, "weekly", "Weekly", "isWeekly", "IsWeekly", "weeklies", "Weeklies"), + "MktCap": get_value(api_record, "mktCap", "MktCap", "marketCap", "MarketCap", "marketCapitalization", "MarketCapitalization"), + "OI": get_value(api_record, "oi", "OI", "openInterest", "OpenInterest", "openInt", "OpenInt") + } + + return mapped + +# ───────────────────────────────────────────── +# Normalization + JSON safety +# ───────────────────────────────────────────── +WEIRD_STR = {"inf","+inf","-inf","infinity","+infinity","-infinity","∞","+∞","-∞", + "nan","-nan","NaN","N/A","NA","NULL","null",""} + +def _coerce_weird_numbers(df: pd.DataFrame) -> pd.DataFrame: + df = df.copy() + for c in df.columns: + if df[c].dtype == object: + df[c] = df[c].replace(list(WEIRD_STR), np.nan) + return df + +def _normalize_for_table(df: pd.DataFrame, table: str) -> pd.DataFrame: + """Rename/select columns to match the DB style of this table.""" + style = TABLE_STYLE.get(table, "snake").lower() + df = df.copy() + df.columns = [c.strip() for c in df.columns] + + if table in ("OptionsFlow","OptionsFlow_monthly"): + if style == "camel": + # Ensure CamelCase headers, no renaming needed if file already CamelCase + # If file is snake, map to Camel + lower_cols = {c for c in df.columns if c.islower()} + if lower_cols: + df = df.rename(columns=MAP_OPTIONSFLOW_CAMEL) + for c in EXPECTED_OPTIONSFLOW_CAMEL: + if c not in df.columns: df[c] = None + df = df[EXPECTED_OPTIONSFLOW_CAMEL] + else: + # snake_case target + # If file CamelCase, map to snake + has_upper = any(any(ch.isupper() for ch in c) for c in df.columns) + if has_upper: + df = df.rename(columns=MAP_OPTIONSFLOW_SNAKE) + for c in EXPECTED_OPTIONSFLOW_SNAKE: + if c not in df.columns: df[c] = None + df = df[EXPECTED_OPTIONSFLOW_SNAKE] + + elif table in ("AlertStream","AlertStream_monthly"): + # DB is snake_case per your screenshot + # If file CamelCase, map to snake + has_upper = any(any(ch.isupper() for ch in c) for c in df.columns) + if has_upper: + df = df.rename(columns=MAP_ALERTSTREAM_SNAKE) + for c in EXPECTED_ALERTSTREAM_SNAKE: + if c not in df.columns: df[c] = None + df = df[EXPECTED_ALERTSTREAM_SNAKE] + + # Standardize any *date columns → YYYY-MM-DD* strings + for col in [c for c in df.columns if c.lower().endswith("date")]: + df[col] = pd.to_datetime(df[col], errors="coerce").dt.strftime("%Y-%m-%d") + + return df + +def _json_safe(df: pd.DataFrame) -> pd.DataFrame: + df = df.copy() + # numeric: drop non-finite + for c in df.columns: + if pd.api.types.is_numeric_dtype(df[c]): + s = pd.to_numeric(df[c], errors="coerce") + s[~np.isfinite(s)] = np.nan + df[c] = s + # datetimes -> strings + for c in df.columns: + if pd.api.types.is_datetime64_any_dtype(df[c]): + df[c] = df[c].dt.strftime("%Y-%m-%d %H:%M:%S") + # NA -> None + return df.astype(object).where(pd.notnull(df), None) + +def _row_hash_from_series(s: pd.Series) -> str: + vals=[] + for _,v in s.items(): + if v is None or (isinstance(v,float) and pd.isna(v)): vals.append("NULL") + elif isinstance(v,str): vals.append(v.strip()) + else: vals.append(str(v)) + return hashlib.sha1("\x1f".join(vals).encode("utf-8","ignore")).hexdigest() + +def _df_prepare_for_postgres(df: pd.DataFrame) -> pd.DataFrame: + if df.empty: return df + df = _json_safe(df) + df["row_hash"] = df.apply(_row_hash_from_series, axis=1) + return df.drop_duplicates(subset=["row_hash"], keep="first").reset_index(drop=True) + +def _extract_missing_col_from_error(msg: str) -> Optional[str]: + """Extract missing column name from PostgreSQL error messages.""" + patterns = [ + r"column \"([^\"]+)\" does not exist", + r"Could not find the '([^']+)' column", + ] + for pattern in patterns: + m = re.search(pattern, msg, re.IGNORECASE) + if m: + return m.group(1) + return None + +# ───────────────────────────────────────────── +# PostgreSQL upload (quiet, chunked) +# ───────────────────────────────────────────── +def _ensure_table_exists(conn, table_name: str, columns: list): + """Ensure table exists with row_hash column and unique constraint.""" + cur = conn.cursor() + try: + # Check if table exists + cur.execute(""" + SELECT EXISTS ( + SELECT FROM information_schema.tables + WHERE table_schema = current_schema() + AND table_name = %s + ) + """, (table_name,)) + exists = cur.fetchone()[0] + + if not exists: + # Create table with all columns as text initially (we'll let PostgreSQL infer types) + # For now, we'll create a basic structure - actual schema should match your data + col_defs = ", ".join([f'"{col}" TEXT' for col in columns if col != "row_hash"]) + col_defs += ', "row_hash" TEXT UNIQUE' + + cur.execute(f'CREATE TABLE IF NOT EXISTS "{table_name}" ({col_defs})') + conn.commit() + else: + # Ensure row_hash column and unique constraint exist + cur.execute(""" + SELECT column_name FROM information_schema.columns + WHERE table_schema = current_schema() + AND table_name = %s AND column_name = 'row_hash' + """, (table_name,)) + if not cur.fetchone(): + cur.execute(f'ALTER TABLE "{table_name}" ADD COLUMN IF NOT EXISTS "row_hash" TEXT') + conn.commit() + + # Check for unique constraint on row_hash + cur.execute(""" + SELECT constraint_name FROM information_schema.table_constraints + WHERE table_schema = current_schema() + AND table_name = %s + AND constraint_type = 'UNIQUE' + AND constraint_name LIKE %s + """, (table_name, f'%{table_name}%row_hash%')) + if not cur.fetchone(): + try: + cur.execute(f'CREATE UNIQUE INDEX IF NOT EXISTS "{table_name}_row_hash_idx" ON "{table_name}" ("row_hash")') + conn.commit() + except Exception: + conn.rollback() + finally: + cur.close() + +def _upsert_slice(conn, tname: str, rows: list, columns: list): + """Upsert a slice of rows using PostgreSQL INSERT ... ON CONFLICT.""" + if not rows: + return + + cur = conn.cursor() + try: + # Ensure table exists + _ensure_table_exists(conn, tname, columns) + + # Build INSERT ... ON CONFLICT statement + cols_quoted = ", ".join([f'"{col}"' for col in columns]) + updates = ", ".join([f'"{col}" = EXCLUDED."{col}"' for col in columns if col != "row_hash"]) + + # Build the base query template for execute_values + base_query = f'INSERT INTO "{tname}" ({cols_quoted}) VALUES %s' + if updates: + conflict_query = f'{base_query} ON CONFLICT ("row_hash") DO UPDATE SET {updates}' + else: + conflict_query = f'{base_query} ON CONFLICT ("row_hash") DO NOTHING' + + # Prepare values as list of tuples + values = [tuple(row.get(col) for col in columns) for row in rows] + + execute_values(cur, conflict_query, values, template=None, page_size=len(rows)) + conn.commit() + except Exception: + conn.rollback() + raise + finally: + cur.close() + +def _postgres_upsert(table: str, df: pd.DataFrame): + if df.empty: + return + + tname = table + total = len(df) + columns = [col for col in df.columns if col != "row_hash"] + ["row_hash"] + + ranges = [(0, min(UPSERT_CHUNK,total))] if SINGLE_CHUNK else \ + [(i, min(i+UPSERT_CHUNK,total)) for i in range(0,total,UPSERT_CHUNK)] + + if not PRINT_PROGRESS: + lbl = "(single-chunk)" if SINGLE_CHUNK else f"chunk_size={UPSERT_CHUNK}" + print_flush(f"{_ts()} | ☁️ Starting upsert → [{tname}] rows={total:,} {lbl}") + + sent = 0 + conn = None + + try: + for start, end in ranges: + part = df.iloc[start:end].copy() + if "row_hash" in part.columns: + part = part.drop_duplicates(subset=["row_hash"], keep="first") + + # Ensure all columns are present + for col in columns: + if col not in part.columns: + part[col] = None + + rows = part[columns].to_dict(orient="records") + + tried_prune = False + for attempt in range(1, MAX_RETRIES + 1): + try: + if conn is None or conn.closed: + conn = _pg_conn() + _upsert_slice(conn, tname, rows, columns) + break + except Exception as e: + missing = _extract_missing_col_from_error(str(e)) + if (missing is not None) and (missing in part.columns) and (not tried_prune): + if PRINT_PRUNE_LOGS: + print_flush(f"{_ts()} | ⚠️ [{tname}] pruning missing column '{missing}'") + part = part.drop(columns=[missing]) + columns = [c for c in columns if c != missing] + rows = part[columns].to_dict(orient="records") + tried_prune = True + continue + if attempt == MAX_RETRIES: + raise + time.sleep(1.0 * attempt) + if conn and not conn.closed: + conn.close() + conn = None + + sent += len(part) + if PRINT_PROGRESS: + print_flush(f"{_ts()} | ☁️ [{tname}] {sent:,}/{total:,}") + + if not PRINT_PROGRESS: + print_flush(f"{_ts()} | ☁️ [{tname}] done {sent:,}/{total:,}") + finally: + if conn and not conn.closed: + conn.close() + +def _postgres_replace(table: str, df: pd.DataFrame): + """Replace all data in table (delete then insert).""" + conn = _pg_conn() + try: + cur = conn.cursor() + cur.execute(f'DELETE FROM "{table}"') + conn.commit() + cur.close() + except Exception: + conn.rollback() + # Table might not exist, that's okay + finally: + conn.close() + + _postgres_upsert(table, df) + +def _load_to_postgres(df: pd.DataFrame, table_name: str, _source_path: str = ""): + ndf = _normalize_for_table(df, table_name) + if ndf.empty: + print_flush(f"{_ts()} | ☁️ Empty after normalize. Skipping PostgreSQL [{table_name}]") + return + + ndf = _df_prepare_for_postgres(ndf) + + if INCREMENTAL: + _postgres_upsert(table_name, ndf) + print_flush(f"{_ts()} | ☁️✅ Upserted {len(ndf):,} rows → PostgreSQL [{table_name}]") + else: + _postgres_replace(table_name, ndf) + print_flush(f"{_ts()} | ☁️✅ Replaced table with {len(ndf):,} rows → PostgreSQL [{table_name}]") + +# ───────────────────────────────────────────── +# Orchestrator +# ───────────────────────────────────────────── +def _load_records_to_databases(records: List[Dict], table_name: str): + """Load records from API into both SQLite and PostgreSQL.""" + if not records: + print_flush(f"{_ts()} | ⚠️ No records to process") + return + + # Convert records to DataFrame + df = pd.DataFrame(records) + df = _coerce_weird_numbers(df) + + if WRITE_POSTGRES: + try: + _load_to_postgres(df, table_name) + except Exception as e: + print_flush(f"{_ts()} | ❌ (PostgreSQL) Error: {e}") + +# ───────────────────────────────────────────── +# Main +# ───────────────────────────────────────────── +def sync_blackbox_flow(): + """Main sync function.""" + print_flush(f"{_ts()} | 🚀 Starting BlackBox Stocks flow data sync...\n") + + try: + # Parse command line arguments + parser = argparse.ArgumentParser(description="Sync BlackBox Stocks options flow data to databases") + parser.add_argument("--start-date", type=str, help="Start date (YYYY-MM-DD)") + parser.add_argument("--end-date", type=str, help="End date (YYYY-MM-DD)") + parser.add_argument("--limit", type=int, help="Maximum number of records to fetch") + parser.add_argument("--count", type=int, help="Maximum number of records to fetch (alias for --limit)") + parser.add_argument("--symbol", type=str, help="Filter by specific symbol") + parser.add_argument("--table", type=str, default="OptionsFlow_monthly", help="Target table name (default: OptionsFlow_monthly)") + + args = parser.parse_args() + + options = {} + if args.start_date: + options["startDate"] = args.start_date + if args.end_date: + options["endDate"] = args.end_date + if args.limit: + options["count"] = args.limit + elif args.count: + options["count"] = args.count + if args.symbol: + options["symbol"] = args.symbol + + table_name = args.table + + # Default to today if no dates provided + if not options.get("startDate") and not options.get("endDate"): + today = date.today().isoformat() + options["startDate"] = today + options["endDate"] = today + print_flush(f"{_ts()} | 📅 No date range specified, using today: {today}") + + print_flush(f"{_ts()} | 📥 Fetching flow data from BlackBox API...") + print_flush(f"{_ts()} | Options: {options}") + + # Fetch data from API + api_records = fetch_blackbox_flow(options) + print_flush(f"{_ts()} | ✅ Fetched {len(api_records)} records from API") + + if len(api_records) == 0: + print_flush(f"{_ts()} | ⚠️ No records returned from API") + return + + # Log sample record + if len(api_records) > 0: + print_flush(f"\n{_ts()} | 📋 Sample API record structure:") + print_flush(json.dumps(api_records[0], indent=2, default=str)) + + # Map API records to database schema + print_flush(f"\n{_ts()} | 🔄 Mapping records to database schema...") + mapped_records = [map_blackbox_to_database(record) for record in api_records] + print_flush(f"{_ts()} | ✅ Mapped {len(mapped_records)} records") + + # Log sample mapped record + if len(mapped_records) > 0: + print_flush(f"\n{_ts()} | 📋 Sample mapped record:") + print_flush(json.dumps(mapped_records[0], indent=2, default=str)) + + # Insert into databases + print_flush(f"\n{_ts()} | 💾 Inserting records into databases...") + _load_records_to_databases(mapped_records, table_name) + + print_flush(f"\n{_ts()} | ✅ Successfully synced {len(mapped_records)} records") + + # Summary + print_flush(f"\n{_ts()} | 📊 Sync Summary:") + print_flush(f"{_ts()} | Fetched from API: {len(api_records)} records") + print_flush(f"{_ts()} | Inserted into DB: {len(mapped_records)} records") + + # Get total count in PostgreSQL database + if WRITE_POSTGRES: + try: + conn = _pg_conn() + cur = conn.cursor() + cur.execute(f'SELECT COUNT(*) FROM "{table_name}"') + total_count = cur.fetchone()[0] + print_flush(f"{_ts()} | Total records in PostgreSQL DB: {total_count}") + conn.close() + except Exception as e: + print_flush(f"{_ts()} | ⚠️ Could not get PostgreSQL count: {e}") + + except Exception as e: + print_flush(f"\n{_ts()} | ❌ Sync failed: {e}") + import traceback + traceback.print_exc() + raise + +if __name__ == "__main__": + if WRITE_POSTGRES: + print_flush(f"{_ts()} | ☁️ PostgreSQL: {POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB} " + f"(schema={POSTGRES_SCHEMA}) | INCREMENTAL={INCREMENTAL} | chunk={UPSERT_CHUNK}") + sync_blackbox_flow()