311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""
|
|
FastAPI service for options flow processing
|
|
Replaces complex SQL with Python/pandas logic
|
|
"""
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
from typing import Optional, List
|
|
from datetime import datetime, timedelta
|
|
import pandas as pd
|
|
import asyncpg
|
|
|
|
from db import get_pool, close_pool
|
|
from services.options_flow_processor import OptionsFlowProcessor
|
|
from services.price_context import PriceContextService
|
|
from services.alert_service import AlertService
|
|
from services.output_formatter import OutputFormatter
|
|
from utils.logger import logger
|
|
from utils.error_handler import handle_processing_error
|
|
|
|
app = FastAPI(title="Options Flow Processing Service", version="1.0.0")
|
|
|
|
# CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # Configure appropriately for production
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
class OptionsFlowRequest(BaseModel):
|
|
start_date: Optional[str] = None
|
|
end_date: Optional[str] = None
|
|
min_premium: Optional[float] = 80000
|
|
tol_pct: Optional[float] = 0.20
|
|
|
|
|
|
class OptionsFlowResponse(BaseModel):
|
|
success: bool
|
|
data: List[dict]
|
|
count: int
|
|
timestamp: str
|
|
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
"""Initialize database pool on startup"""
|
|
try:
|
|
logger.info("Initializing database connection pool...")
|
|
pool = await get_pool()
|
|
# Test the connection with a quick query
|
|
async with pool.acquire() as conn:
|
|
await conn.fetchval("SELECT 1")
|
|
logger.info("✅ Database connection pool initialized successfully")
|
|
except Exception as e:
|
|
logger.error(f"⚠️ Failed to initialize database pool on startup: {str(e)}")
|
|
logger.warning("Service will start but database operations may fail. Connection will be retried on first request.")
|
|
# Don't raise - allow service to start and retry on first request
|
|
# This makes the service more resilient to temporary DB issues
|
|
|
|
|
|
@app.on_event("shutdown")
|
|
async def shutdown():
|
|
"""Close database pool on shutdown"""
|
|
await close_pool()
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
"""Health check endpoint"""
|
|
try:
|
|
pool = await get_pool()
|
|
async with pool.acquire() as conn:
|
|
await conn.fetchval("SELECT 1")
|
|
return {"status": "healthy", "service": "options-flow-processor"}
|
|
except Exception as e:
|
|
return {"status": "unhealthy", "error": str(e)}
|
|
|
|
|
|
@app.get("/api/options-flow", response_model=OptionsFlowResponse)
|
|
async def get_options_flow(
|
|
start_date: Optional[str] = Query(None, description="Start date (YYYY-MM-DD)"),
|
|
end_date: Optional[str] = Query(None, description="End date (YYYY-MM-DD)"),
|
|
min_premium: Optional[float] = Query(80000, description="Minimum premium filter"),
|
|
tol_pct: Optional[float] = Query(0.20, description="Tape alignment tolerance")
|
|
):
|
|
"""
|
|
Get processed options flow data
|
|
Replaces the complex SQL query with Python processing
|
|
"""
|
|
try:
|
|
logger.info(f"Options flow request: start={start_date}, end={end_date}, min_premium={min_premium}")
|
|
pool = await get_pool()
|
|
|
|
# Default dates (only if not provided)
|
|
if not start_date:
|
|
start_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
|
logger.info(f"No start_date provided, using default: {start_date}")
|
|
if not end_date:
|
|
end_date = datetime.now().strftime('%Y-%m-%d')
|
|
logger.info(f"No end_date provided, using default: {end_date}")
|
|
|
|
logger.info(f"Processing with date range: {start_date} to {end_date}")
|
|
start_dt = datetime.strptime(start_date, '%Y-%m-%d')
|
|
end_dt = datetime.strptime(end_date, '%Y-%m-%d')
|
|
|
|
# Load raw options flow data (with timeout handling)
|
|
try:
|
|
async with pool.acquire() as conn:
|
|
query = """
|
|
SELECT *
|
|
FROM "OptionsFlow_monthly"
|
|
WHERE "Premium" IS NOT NULL
|
|
AND TRIM("Premium"::text) <> ''
|
|
AND "StockEtf" = 'STOCK'
|
|
AND "Symbol" NOT IN ('TSLA', 'NVDA')
|
|
"""
|
|
rows = await conn.fetch(query)
|
|
except Exception as e:
|
|
logger.error(f"Database query error: {type(e).__name__} - {str(e)}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Database query failed: {str(e)}"
|
|
)
|
|
|
|
if not rows:
|
|
return OptionsFlowResponse(
|
|
success=True,
|
|
data=[],
|
|
count=0,
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
# Convert to DataFrame
|
|
df = pd.DataFrame([dict(row) for row in rows])
|
|
|
|
# Process with Python service
|
|
processor = OptionsFlowProcessor(tol_pct=tol_pct)
|
|
df_processed = processor.process(df, start_dt, end_dt)
|
|
|
|
# Enrich with price context (optimized batch queries)
|
|
price_service = PriceContextService(pool)
|
|
df_with_prices = await price_service.enrich_flow_with_prices(df_processed, pool)
|
|
|
|
# Match alerts (batch processing)
|
|
alert_service = AlertService(pool)
|
|
df_final = await alert_service.match_alerts_to_flows(df_with_prices)
|
|
|
|
# Recalculate rocket score with price context and alerts
|
|
df_final = processor.process_rocket_score(df_final)
|
|
|
|
# Check if DataFrame is empty before filtering
|
|
if df_final.empty:
|
|
logger.warning("No data after processing, returning empty result")
|
|
return OptionsFlowResponse(
|
|
success=True,
|
|
data=[],
|
|
count=0,
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
# Filter by minimum premium and badge requirements (matching SQL WHERE clause)
|
|
logger.info(f"📊 Before filtering: {len(df_final)} rows")
|
|
|
|
# Only filter if columns exist
|
|
if 'premium_num' in df_final.columns:
|
|
before_premium = len(df_final)
|
|
df_final = df_final[df_final['premium_num'] > min_premium].copy()
|
|
after_premium = len(df_final)
|
|
logger.info(f"📊 After premium filter (>${min_premium:,.0f}): {after_premium} rows (removed {before_premium - after_premium})")
|
|
else:
|
|
logger.warning("⚠️ premium_num column not found, skipping premium filter")
|
|
|
|
if df_final.empty:
|
|
logger.warning("⚠️ No data after premium filter")
|
|
return OptionsFlowResponse(
|
|
success=True,
|
|
data=[],
|
|
count=0,
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
# Filter by badge requirements (only if columns exist)
|
|
if 'badge_round' in df_final.columns and 'badge_more' in df_final.columns:
|
|
before_badges = len(df_final)
|
|
df_final = df_final[
|
|
(df_final['badge_round'].isin(['🟢', '🔴'])) &
|
|
(df_final['badge_more'].str.contains('💎', na=False)) &
|
|
(df_final['badge_more'].str.contains('⭐', na=False))
|
|
].copy()
|
|
after_badges = len(df_final)
|
|
logger.info(f"📊 After badge filter (🟢/🔴 + 💎 + ⭐): {after_badges} rows (removed {before_badges - after_badges})")
|
|
else:
|
|
logger.warning("⚠️ badge_round or badge_more columns not found, skipping badge filter")
|
|
|
|
if df_final.empty:
|
|
logger.warning("⚠️ No data after badge filter")
|
|
return OptionsFlowResponse(
|
|
success=True,
|
|
data=[],
|
|
count=0,
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
# Additional direction filter (only if columns exist)
|
|
if 'direction' in df_final.columns and 'badge_round' in df_final.columns and 'bull_total' in df_final.columns and 'bear_total' in df_final.columns:
|
|
before_direction = len(df_final)
|
|
df_final = df_final[
|
|
((df_final['direction'] == 'BULL') &
|
|
(df_final['badge_round'] == '🟢') &
|
|
((df_final['bull_total'] - df_final['bear_total']) > 0)) |
|
|
((df_final['direction'] == 'BEAR') &
|
|
(df_final['badge_round'] == '🔴') &
|
|
((df_final['bull_total'] - df_final['bear_total']) < 0))
|
|
].copy()
|
|
after_direction = len(df_final)
|
|
logger.info(f"📊 After direction/net premium filter: {after_direction} rows (removed {before_direction - after_direction})")
|
|
else:
|
|
logger.warning("⚠️ Required columns for direction filter not found, skipping")
|
|
|
|
# Sort by timestamp descending
|
|
df_final = df_final.sort_values(['flow_ts_utc', 'rid'], ascending=[False, False])
|
|
|
|
# Format output to match SQL format
|
|
df_final = OutputFormatter.format_final_output(df_final)
|
|
|
|
# Convert DataFrame to list of dicts
|
|
result_data = df_final.to_dict('records')
|
|
|
|
# Format dates and handle NaN values
|
|
for record in result_data:
|
|
# Convert datetime objects to strings
|
|
for key, value in record.items():
|
|
if isinstance(value, datetime):
|
|
record[key] = value.isoformat()
|
|
elif pd.isna(value):
|
|
record[key] = None
|
|
elif isinstance(value, pd.Timestamp):
|
|
# Check if it's NaT (Not a Time)
|
|
if pd.isna(value):
|
|
record[key] = None
|
|
else:
|
|
record[key] = value.isoformat()
|
|
|
|
return OptionsFlowResponse(
|
|
success=True,
|
|
data=result_data,
|
|
count=len(result_data),
|
|
timestamp=datetime.now().isoformat()
|
|
)
|
|
|
|
except Exception as e:
|
|
error_info = handle_processing_error(
|
|
e,
|
|
context={
|
|
'start_date': start_date,
|
|
'end_date': end_date,
|
|
'min_premium': min_premium
|
|
},
|
|
raise_error=False
|
|
)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=error_info.get('error_message', str(e)) if isinstance(error_info, dict) else str(e)
|
|
)
|
|
|
|
|
|
@app.get("/api/options-flow/stats")
|
|
async def get_flow_stats(
|
|
symbol: Optional[str] = Query(None, description="Symbol to get stats for")
|
|
):
|
|
"""Get flow statistics"""
|
|
try:
|
|
pool = await get_pool()
|
|
|
|
query = """
|
|
SELECT
|
|
symbol,
|
|
COUNT(*) as total_trades,
|
|
SUM(premium_num) as total_premium,
|
|
SUM(CASE WHEN cp_norm = 'CALL' THEN vol_num ELSE 0 END) as call_volume,
|
|
SUM(CASE WHEN cp_norm = 'PUT' THEN vol_num ELSE 0 END) as put_volume
|
|
FROM processed_options_flow
|
|
"""
|
|
|
|
params = []
|
|
if symbol:
|
|
query += " WHERE symbol_norm = $1"
|
|
params.append(symbol.upper())
|
|
|
|
query += " GROUP BY symbol"
|
|
|
|
async with pool.acquire() as conn:
|
|
rows = await conn.fetch(query, *params)
|
|
|
|
return {
|
|
"success": True,
|
|
"data": [dict(row) for row in rows]
|
|
}
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8010)
|
|
|