lewismac
9db0c4ca80
- Add GET /api/indicators/coverage/{symbol}/{interval} to check coverage
- Add GET /api/indicators/coverage/all for system-wide coverage status
- Add POST /api/indicators/backfill/{symbol}/{interval} to backfill missing indicators
- Add POST /api/indicators/backfill-all for bulk backfill operations
- Add GET /api/indicators/missing/{symbol}/{interval} to list incomplete records
- Add GET /api/indicators/summary for aggregate coverage statistics
- Support configurable batch_size and min_coverage_threshold parameters
- Return detailed results including before/after coverage percentages
- Provide summary statistics with coverage ranges and lowest coverage pairs
- Enable UI integration for monitoring and managing indicator completeness
These endpoints expose the db.py indicator coverage methods through the
web API, allowing users to monitor and maintain 100% technical indicator
coverage across all trading pairs via the web interface.
1047 lines
45 KiB
Python
1047 lines
45 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""
|
|
|
|
ui_routes.py - API Endpoints and Route Handlers
|
|
|
|
Defines all FastAPI routes and business logic for API endpoints
|
|
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Dict, Any, Optional, List
|
|
from fastapi import HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
from dotenv import set_key, dotenv_values
|
|
|
|
# UI and models
|
|
from ui_models import (
|
|
TradingPairConfig,
|
|
TradingPairAdd,
|
|
BulkDownloadRequest,
|
|
GapFillRequest,
|
|
ConfigUpdate,
|
|
EnvVarUpdate,
|
|
ChartDataRequest,
|
|
AutoGapFillRequest,
|
|
GapDetectionRequest,
|
|
serialize_for_json,
|
|
)
|
|
|
|
from ui_template_dashboard import get_dashboard_html
|
|
from ui_template_config import get_config_html
|
|
from ui_state import get_current_status
|
|
from utils import load_config, save_config, validate_symbol, reload_env_vars
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _tz_aware(dt: datetime) -> datetime:
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
def _ok(data: Any, status: str = "success", http_status: int = 200) -> JSONResponse:
|
|
return JSONResponse(content={"status": status, "data": serialize_for_json(data)}, status_code=http_status)
|
|
|
|
|
|
def _err(message: str, http_status: int = 500, extra: Optional[Dict[str, Any]] = None) -> JSONResponse:
|
|
payload = {"status": "error", "message": message}
|
|
if extra:
|
|
payload.update(extra)
|
|
return JSONResponse(content=payload, status_code=http_status)
|
|
|
|
|
|
class APIRoutes:
|
|
"""Encapsulates all API route handlers"""
|
|
|
|
def __init__(self, app, db_manager, data_collector, config, state_manager):
|
|
self.app = app
|
|
self.db_manager = db_manager
|
|
self.data_collector = data_collector
|
|
self.config = config
|
|
self.state_manager = state_manager
|
|
|
|
# Register all routes
|
|
self._register_routes()
|
|
|
|
def _register_routes(self):
|
|
"""Register all API routes"""
|
|
|
|
# ---------------------------
|
|
# Pages
|
|
# ---------------------------
|
|
|
|
@self.app.get("/", response_class=HTMLResponse)
|
|
async def dashboard():
|
|
"""Serve the main dashboard"""
|
|
return get_dashboard_html()
|
|
|
|
@self.app.get("/config", response_class=HTMLResponse)
|
|
async def config_page():
|
|
"""Serve the configuration management page"""
|
|
return get_config_html()
|
|
|
|
@self.app.get("/gaps", response_class=HTMLResponse)
|
|
async def gaps_page():
|
|
"""Serve the gap monitoring page"""
|
|
from ui_template_gaps import get_gaps_monitoring_html
|
|
return get_gaps_monitoring_html()
|
|
|
|
# ---------------------------
|
|
# Status
|
|
# ---------------------------
|
|
|
|
@self.app.get("/api/stats")
|
|
async def get_stats():
|
|
"""Get current system statistics"""
|
|
try:
|
|
status = await get_current_status(self.db_manager, self.data_collector, self.config)
|
|
return JSONResponse(content=serialize_for_json(status))
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting stats: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
# ---------------------------
|
|
# Indicator Coverage (NEW)
|
|
# ---------------------------
|
|
|
|
@self.app.get("/api/indicators/coverage/{symbol}/{interval}")
|
|
async def get_indicator_coverage(symbol: str, interval: str):
|
|
"""Get technical indicator coverage status for a symbol/interval"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
coverage = await self.db_manager.check_indicator_coverage(symbol.upper(), interval)
|
|
return _ok(coverage)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking indicator coverage: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/indicators/coverage/all")
|
|
async def get_all_indicator_coverage():
|
|
"""Get indicator coverage status for all symbol/interval combinations"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
logger.info("Fetching indicator coverage for all pairs")
|
|
coverage_list = await self.db_manager.get_all_indicator_coverage_status()
|
|
|
|
logger.info(f"Retrieved coverage for {len(coverage_list)} combinations")
|
|
return _ok(coverage_list)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting all indicator coverage: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/indicators/backfill/{symbol}/{interval}")
|
|
async def backfill_indicators(symbol: str, interval: str, request: Request):
|
|
"""Backfill missing technical indicators for a symbol/interval"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
body = await request.json() if request.headers.get('content-type') == 'application/json' else {}
|
|
batch_size = body.get('batch_size', 200)
|
|
|
|
logger.info(f"Starting indicator backfill for {symbol} {interval} (batch_size={batch_size})")
|
|
|
|
result = await self.db_manager.backfill_missing_indicators(
|
|
symbol.upper(), interval, batch_size=batch_size
|
|
)
|
|
|
|
logger.info(f"Indicator backfill completed: {result}")
|
|
return _ok(result)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error backfilling indicators: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/indicators/backfill-all")
|
|
async def backfill_all_indicators(request: Request):
|
|
"""Backfill missing indicators for all symbol/interval combinations with incomplete coverage"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
body = await request.json() if request.headers.get('content-type') == 'application/json' else {}
|
|
batch_size = body.get('batch_size', 200)
|
|
min_coverage_threshold = body.get('min_coverage_threshold', 99.9)
|
|
|
|
logger.info("Starting indicator backfill for all pairs with incomplete coverage")
|
|
|
|
# Get all coverage status
|
|
coverage_list = await self.db_manager.get_all_indicator_coverage_status()
|
|
|
|
# Filter to only those needing backfill
|
|
needs_backfill = [
|
|
c for c in coverage_list
|
|
if c['coverage_percent'] < min_coverage_threshold
|
|
]
|
|
|
|
logger.info(f"Found {len(needs_backfill)} combinations needing backfill")
|
|
|
|
results = []
|
|
for coverage in needs_backfill:
|
|
symbol = coverage['symbol']
|
|
interval = coverage['interval']
|
|
|
|
try:
|
|
result = await self.db_manager.backfill_missing_indicators(
|
|
symbol, interval, batch_size=batch_size
|
|
)
|
|
results.append({
|
|
'symbol': symbol,
|
|
'interval': interval,
|
|
'status': result['status'],
|
|
'coverage_before': result.get('coverage_before', 0),
|
|
'coverage_after': result.get('coverage_after', 0),
|
|
'indicators_added': result.get('indicators_added', 0)
|
|
})
|
|
|
|
logger.info(
|
|
f"Backfilled {symbol} {interval}: "
|
|
f"{result.get('coverage_before', 0):.2f}% → {result.get('coverage_after', 0):.2f}%"
|
|
)
|
|
|
|
# Small delay to avoid overwhelming the database
|
|
await asyncio.sleep(0.5)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error backfilling {symbol} {interval}: {e}")
|
|
results.append({
|
|
'symbol': symbol,
|
|
'interval': interval,
|
|
'status': 'error',
|
|
'error': str(e)
|
|
})
|
|
|
|
summary = {
|
|
'total_checked': len(coverage_list),
|
|
'needed_backfill': len(needs_backfill),
|
|
'processed': len(results),
|
|
'successful': len([r for r in results if r.get('status') == 'success']),
|
|
'failed': len([r for r in results if r.get('status') == 'error']),
|
|
'results': results
|
|
}
|
|
|
|
logger.info(f"Bulk indicator backfill complete: {summary['successful']}/{summary['processed']} successful")
|
|
return _ok(summary)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in bulk indicator backfill: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/indicators/missing/{symbol}/{interval}")
|
|
async def get_missing_indicators(symbol: str, interval: str):
|
|
"""Get list of OHLCV records missing technical indicators"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
missing = await self.db_manager.get_ohlcv_missing_indicators(
|
|
symbol.upper(), interval, limit=100
|
|
)
|
|
|
|
return _ok({
|
|
'symbol': symbol.upper(),
|
|
'interval': interval,
|
|
'missing_count': len(missing),
|
|
'missing_records': missing
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting missing indicators: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/indicators/summary")
|
|
async def get_indicators_summary():
|
|
"""Get summary of indicator coverage across all symbols"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
coverage_list = await self.db_manager.get_all_indicator_coverage_status()
|
|
|
|
# Calculate summary statistics
|
|
total_combinations = len(coverage_list)
|
|
complete_combinations = len([c for c in coverage_list if c['coverage_percent'] >= 99.9])
|
|
incomplete_combinations = total_combinations - complete_combinations
|
|
|
|
avg_coverage = sum(c['coverage_percent'] for c in coverage_list) / total_combinations if total_combinations > 0 else 0
|
|
|
|
# Group by coverage ranges
|
|
coverage_ranges = {
|
|
'100%': len([c for c in coverage_list if c['coverage_percent'] >= 99.9]),
|
|
'90-99%': len([c for c in coverage_list if 90 <= c['coverage_percent'] < 99.9]),
|
|
'50-89%': len([c for c in coverage_list if 50 <= c['coverage_percent'] < 90]),
|
|
'0-49%': len([c for c in coverage_list if c['coverage_percent'] < 50]),
|
|
}
|
|
|
|
summary = {
|
|
'total_combinations': total_combinations,
|
|
'complete_combinations': complete_combinations,
|
|
'incomplete_combinations': incomplete_combinations,
|
|
'average_coverage': round(avg_coverage, 2),
|
|
'coverage_ranges': coverage_ranges,
|
|
'lowest_coverage': sorted(coverage_list, key=lambda x: x['coverage_percent'])[:10] if coverage_list else []
|
|
}
|
|
|
|
return _ok(summary)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting indicators summary: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
# ---------------------------
|
|
# Gaps and Coverage
|
|
# ---------------------------
|
|
|
|
@self.app.get("/api/gaps/all-pairs")
|
|
async def get_all_pairs_gaps():
|
|
"""Get gap status for all trading pairs"""
|
|
try:
|
|
if not self.db_manager:
|
|
logger.error("Database manager not initialized")
|
|
return _err("Database not initialized", 500)
|
|
|
|
logger.info("Fetching gap status for all pairs")
|
|
status = await self.db_manager.get_all_pairs_gap_status()
|
|
|
|
logger.info(f"Retrieved gap status for {len(status)} pair-interval combinations")
|
|
return _ok(status)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting all pairs gaps: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/gaps/details/{symbol}/{interval}")
|
|
async def get_gap_details(symbol: str, interval: str):
|
|
"""Get detailed gap information including daily coverage"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
sym = symbol.upper()
|
|
gap_info = await self.db_manager.detect_gaps(sym, interval)
|
|
|
|
end_date = datetime.utcnow()
|
|
start_date = end_date - timedelta(days=90)
|
|
|
|
daily_coverage = await self.db_manager.get_data_coverage_by_day(sym, interval, start_date, end_date)
|
|
|
|
data = {
|
|
"coverage_percent": gap_info.get('coverage', {}).get('coverage_percent', 0),
|
|
"total_records": gap_info.get('coverage', {}).get('total_records', 0),
|
|
"missing_records": gap_info.get('coverage', {}).get('missing_records', 0),
|
|
"gaps": gap_info.get('gaps', []),
|
|
"daily_coverage": daily_coverage.get('daily_coverage', []),
|
|
}
|
|
|
|
return _ok(data)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting gap details: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/gaps/fill-intelligent")
|
|
async def fill_gaps_intelligent(request: Request):
|
|
"""Intelligently fill gaps with multiple attempts and averaging fallback"""
|
|
try:
|
|
body = await request.json()
|
|
symbol = body.get('symbol')
|
|
interval = body.get('interval')
|
|
max_attempts = int(body.get('max_attempts', 3))
|
|
|
|
if not symbol or not interval:
|
|
return _err("Missing symbol or interval", 400)
|
|
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
result = await self.db_manager.fill_gaps_intelligently(symbol.upper(), interval, max_attempts)
|
|
|
|
logger.info(f"Intelligent gap fill completed: {result}")
|
|
return _ok(result)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in intelligent gap fill: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/gaps/prioritized/{symbol}/{interval}")
|
|
async def get_prioritized_gaps(symbol: str, interval: str):
|
|
"""Get gaps sorted by priority (recent and small gaps first)"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
prioritized = await self.db_manager.get_prioritized_gaps(symbol.upper(), interval)
|
|
return _ok(prioritized)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting prioritized gaps: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/gaps/progress/{symbol}/{interval}")
|
|
async def get_gap_progress(symbol: str, interval: str):
|
|
"""Get real-time progress and estimated completion time"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
progress = await self.db_manager.get_gap_fill_progress(symbol.upper(), interval)
|
|
return _ok(progress)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting gap progress: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/gaps/health/{symbol}/{interval}")
|
|
async def get_data_health(symbol: str, interval: str):
|
|
"""Get comprehensive data health analysis"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
health = await self.db_manager.check_data_health(symbol.upper(), interval)
|
|
return _ok(health)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking data health: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/gaps/smart-fill/{symbol}")
|
|
async def smart_fill_gaps(symbol: str):
|
|
"""Intelligently fill gaps starting with highest priority"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
from utils import load_config
|
|
cfg = load_config()
|
|
intervals = cfg.get('collection', {}).get('candle_intervals', ['1m', '5m', '15m', '1h', '4h', '1d'])
|
|
|
|
results: List[Dict[str, Any]] = []
|
|
|
|
for interval in intervals:
|
|
prioritized = await self.db_manager.get_prioritized_gaps(symbol.upper(), interval)
|
|
|
|
if not prioritized:
|
|
continue
|
|
|
|
filled = 0
|
|
for gap in prioritized[:5]: # Only fill top 5 gaps
|
|
if gap.get('missing_candles', 0) <= 100:
|
|
try:
|
|
await self.db_manager.fill_gaps_intelligently(symbol.upper(), interval, max_attempts=3)
|
|
filled += 1
|
|
except Exception as e:
|
|
logger.error(f"Error filling gap: {e}")
|
|
|
|
results.append({'interval': interval, 'gaps_filled': filled, 'total_gaps': len(prioritized)})
|
|
|
|
return JSONResponse(content={
|
|
"status": "success",
|
|
"message": f"Smart fill completed for {symbol}",
|
|
"data": results
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in smart fill: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/gaps/fill")
|
|
async def fill_gaps(request: GapFillRequest):
|
|
"""Fill data gaps"""
|
|
try:
|
|
if not self.data_collector:
|
|
raise HTTPException(status_code=500, detail="Data collector not initialized")
|
|
|
|
gap_start = datetime.fromisoformat(request.gap_start)
|
|
gap_end = datetime.fromisoformat(request.gap_end)
|
|
|
|
gap_start = _tz_aware(gap_start)
|
|
gap_end = _tz_aware(gap_end)
|
|
|
|
await self.data_collector.bulk_download_historical_data(
|
|
request.symbol.upper(),
|
|
gap_start,
|
|
gap_end,
|
|
[request.interval],
|
|
)
|
|
|
|
logger.info(f"Gap filled for {request.symbol} {request.interval}")
|
|
return JSONResponse(content={"status": "success", "message": "Gap filled successfully"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error filling gap: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/gaps/auto-fill")
|
|
async def auto_fill_gaps(request: AutoGapFillRequest):
|
|
"""Automatically fill gaps for a symbol"""
|
|
try:
|
|
if not self.data_collector:
|
|
raise HTTPException(status_code=500, detail="Data collector not initialized")
|
|
|
|
result = await self.data_collector.auto_fill_gaps(
|
|
request.symbol.upper(),
|
|
request.intervals,
|
|
request.fill_genuine_gaps,
|
|
)
|
|
|
|
logger.info(f"Auto gap fill completed for {request.symbol}: {result}")
|
|
return JSONResponse(content={
|
|
"status": "success",
|
|
"message": f"Filled gaps for {request.symbol}",
|
|
"result": serialize_for_json(result)
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in auto gap fill: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/gaps/summary")
|
|
async def get_gaps_summary():
|
|
"""Get summary of all gaps across all symbols"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
summary = await self.db_manager.get_all_gaps_summary()
|
|
return _ok(summary)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting gaps summary: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/gaps/status/{symbol}/{interval}")
|
|
async def get_gap_status(symbol: str, interval: str):
|
|
"""Get gap fill status for a specific symbol/interval"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
status = await self.db_manager.get_gap_fill_status(symbol.upper(), interval)
|
|
return _ok(status)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting gap status: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.get("/api/gaps/{symbol}/{interval}")
|
|
async def detect_gaps(symbol: str, interval: str):
|
|
"""Detect data gaps"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
gaps = await self.db_manager.detect_gaps(symbol.upper(), interval)
|
|
return JSONResponse(content={"status": "success", "gaps": serialize_for_json(gaps)})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error detecting gaps: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/gaps/fill-genuine/{symbol}/{interval}")
|
|
async def fill_genuine_gaps(symbol: str, interval: str):
|
|
"""Fill genuine empty gaps with intelligent averaging"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
gap_config = self.config.get('gap_filling', {})
|
|
max_consecutive = int(gap_config.get('max_consecutive_empty_candles', 5))
|
|
lookback = int(gap_config.get('averaging_lookback_candles', 10))
|
|
|
|
filled_count = await self.db_manager.fill_genuine_gaps_with_averages(
|
|
symbol.upper(), interval, max_consecutive, lookback
|
|
)
|
|
|
|
logger.info(f"Filled {filled_count} genuine gaps for {symbol} {interval}")
|
|
return JSONResponse(
|
|
content={
|
|
"status": "success",
|
|
"message": f"Filled {filled_count} genuine empty candles",
|
|
"filled_count": filled_count,
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error filling genuine gaps: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
# ---------------------------
|
|
# Symbols and Prices
|
|
# ---------------------------
|
|
|
|
@self.app.get("/api/symbols")
|
|
async def get_symbols():
|
|
"""Get list of all available symbols"""
|
|
try:
|
|
if not self.db_manager:
|
|
logger.error("Database manager not initialized")
|
|
return JSONResponse(content={"status": "error", "symbols": []}, status_code=500)
|
|
|
|
symbols = await self.db_manager.get_available_symbols()
|
|
logger.info(f"Retrieved {len(symbols)} symbols from database")
|
|
return JSONResponse(content={"status": "success", "symbols": symbols})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting symbols: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "symbols": []}, status_code=500)
|
|
|
|
@self.app.get("/api/price-trends/{symbol}")
|
|
async def get_price_trends(symbol: str):
|
|
"""Get current price and trend indicators for multiple timeframes"""
|
|
try:
|
|
if not self.db_manager:
|
|
logger.error("Database manager not initialized")
|
|
return _err("Database not initialized", 500)
|
|
|
|
logger.info(f"Price trends request for {symbol}")
|
|
|
|
data = await self.db_manager.get_current_price_and_trends_with_volume(symbol.upper())
|
|
|
|
if not data:
|
|
logger.warning(f"No price data found for {symbol}")
|
|
return _err(f"No data found for {symbol}. Please start data collection first.", 404)
|
|
|
|
pair_config = next((p for p in self.config.get('trading_pairs', []) if p['symbol'] == symbol.upper()), None)
|
|
data['enabled'] = pair_config.get('enabled', False) if pair_config else False
|
|
|
|
logger.info(f"Returning price trends for {symbol}: price={data.get('current_price')}")
|
|
return _ok(data)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting price trends: {e}", exc_info=True)
|
|
return _err(f"Error retrieving price trends: {str(e)}", 500)
|
|
|
|
# ---------------------------
|
|
# Collection control
|
|
# ---------------------------
|
|
|
|
@self.app.post("/api/collection/start")
|
|
async def start_collection():
|
|
"""Start data collection"""
|
|
try:
|
|
if not self.data_collector:
|
|
raise HTTPException(status_code=500, detail="Data collector not initialized")
|
|
|
|
if self.state_manager.get("is_collecting", False):
|
|
return JSONResponse(content={"status": "info", "message": "Collection already running"})
|
|
|
|
await self.data_collector.start_continuous_collection()
|
|
self.state_manager.update(is_collecting=True)
|
|
|
|
logger.info("Collection started via API")
|
|
return JSONResponse(content={"status": "success", "message": "Collection started"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error starting collection: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
@self.app.post("/api/collection/stop")
|
|
async def stop_collection():
|
|
"""Stop data collection"""
|
|
try:
|
|
if not self.data_collector:
|
|
raise HTTPException(status_code=500, detail="Data collector not initialized")
|
|
|
|
if not self.state_manager.get("is_collecting", False):
|
|
return JSONResponse(content={"status": "info", "message": "Collection not running"})
|
|
|
|
await self.data_collector.stop_continuous_collection()
|
|
self.state_manager.update(is_collecting=False)
|
|
|
|
logger.info("Collection stopped via API")
|
|
return JSONResponse(content={"status": "success", "message": "Collection stopped"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error stopping collection: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
# ---------------------------
|
|
# Configuration
|
|
# ---------------------------
|
|
|
|
@self.app.get("/api/config")
|
|
async def get_configuration():
|
|
"""Get current configuration"""
|
|
try:
|
|
cfg = load_config()
|
|
return JSONResponse(content=serialize_for_json(cfg))
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting config: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@self.app.post("/api/config")
|
|
async def update_configuration(request: Request):
|
|
"""Update configuration - accepts raw JSON body"""
|
|
try:
|
|
body = await request.json()
|
|
|
|
logger.info(f"Received config update keys: {list(body.keys())}")
|
|
|
|
current_config = load_config()
|
|
|
|
# Deep merge/replace top-level keys
|
|
for key, value in body.items():
|
|
if key in current_config and isinstance(current_config[key], dict) and isinstance(value, dict):
|
|
current_config[key].update(value)
|
|
else:
|
|
current_config[key] = value
|
|
|
|
save_config(current_config)
|
|
|
|
self.config.clear()
|
|
self.config.update(current_config)
|
|
|
|
logger.info("Configuration updated successfully")
|
|
return JSONResponse(content={"status": "success", "message": "Configuration updated"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating config: {e}", exc_info=True)
|
|
return _err(str(e), 500)
|
|
|
|
@self.app.post("/api/trading-pairs")
|
|
async def add_trading_pair(pair: TradingPairAdd):
|
|
"""Add a new trading pair"""
|
|
try:
|
|
if not validate_symbol(pair.symbol.upper()):
|
|
return JSONResponse(content={"status": "error", "message": "Invalid symbol format"}, status_code=400)
|
|
|
|
cfg = load_config()
|
|
|
|
existing = [p for p in cfg.get('trading_pairs', []) if p['symbol'] == pair.symbol.upper()]
|
|
if existing:
|
|
return JSONResponse(content={"status": "error", "message": "Trading pair already exists"}, status_code=409)
|
|
|
|
record_from_date = pair.record_from_date or cfg.get('collection', {}).get('default_record_from_date', '2020-01-01T00:00:00Z')
|
|
|
|
cfg.setdefault('trading_pairs', []).append({
|
|
'symbol': pair.symbol.upper(),
|
|
'enabled': True,
|
|
'priority': pair.priority,
|
|
'record_from_date': record_from_date,
|
|
})
|
|
|
|
save_config(cfg)
|
|
|
|
self.config.clear()
|
|
self.config.update(cfg)
|
|
|
|
logger.info(f"Added trading pair: {pair.symbol}")
|
|
return JSONResponse(content={"status": "success", "message": f"Added {pair.symbol}"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error adding trading pair: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
@self.app.put("/api/trading-pairs/{symbol}")
|
|
async def update_trading_pair(symbol: str, request: Request):
|
|
"""Update a trading pair's configuration"""
|
|
try:
|
|
update = await request.json()
|
|
logger.info(f"Updating trading pair {symbol}: {update}")
|
|
|
|
cfg = load_config()
|
|
|
|
pair_found = False
|
|
for pair in cfg.get('trading_pairs', []):
|
|
if pair['symbol'] == symbol.upper():
|
|
if 'enabled' in update:
|
|
pair['enabled'] = bool(update['enabled'])
|
|
if 'priority' in update:
|
|
pair['priority'] = int(update['priority'])
|
|
if 'record_from_date' in update:
|
|
pair['record_from_date'] = update['record_from_date']
|
|
|
|
pair_found = True
|
|
break
|
|
|
|
if not pair_found:
|
|
return JSONResponse(content={"status": "error", "message": "Trading pair not found"}, status_code=404)
|
|
|
|
save_config(cfg)
|
|
|
|
self.config.clear()
|
|
self.config.update(cfg)
|
|
|
|
logger.info(f"Updated trading pair: {symbol}")
|
|
return JSONResponse(content={"status": "success", "message": f"Updated {symbol}"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating trading pair: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
@self.app.delete("/api/trading-pairs/{symbol}")
|
|
async def remove_trading_pair(symbol: str):
|
|
"""Remove a trading pair"""
|
|
try:
|
|
cfg = load_config()
|
|
|
|
original_count = len(cfg.get('trading_pairs', []))
|
|
cfg['trading_pairs'] = [p for p in cfg.get('trading_pairs', []) if p['symbol'] != symbol.upper()]
|
|
|
|
if len(cfg['trading_pairs']) == original_count:
|
|
return JSONResponse(content={"status": "error", "message": "Trading pair not found"}, status_code=404)
|
|
|
|
save_config(cfg)
|
|
|
|
self.config.clear()
|
|
self.config.update(cfg)
|
|
|
|
logger.info(f"Removed trading pair: {symbol}")
|
|
return JSONResponse(content={"status": "success", "message": f"Removed {symbol}"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error removing trading pair: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
@self.app.post("/api/indicators/toggle/{indicator_name}")
|
|
async def toggle_indicator(indicator_name: str):
|
|
"""Toggle a technical indicator on/off"""
|
|
try:
|
|
cfg = load_config()
|
|
enabled_indicators = cfg.setdefault('technical_indicators', {}).setdefault('enabled', [])
|
|
|
|
if indicator_name in enabled_indicators:
|
|
enabled_indicators.remove(indicator_name)
|
|
action = "disabled"
|
|
else:
|
|
enabled_indicators.append(indicator_name)
|
|
action = "enabled"
|
|
|
|
save_config(cfg)
|
|
|
|
self.config.clear()
|
|
self.config.update(cfg)
|
|
|
|
logger.info(f"Indicator {indicator_name} {action}")
|
|
return JSONResponse(content={
|
|
"status": "success",
|
|
"message": f"Indicator {indicator_name} {action}",
|
|
"enabled": indicator_name in enabled_indicators
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error toggling indicator: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
@self.app.put("/api/indicators/{indicator_name}/periods")
|
|
async def update_indicator_periods(indicator_name: str, request: Request):
|
|
"""Update periods for a technical indicator"""
|
|
try:
|
|
body = await request.json()
|
|
periods = body.get('periods')
|
|
|
|
if periods is None:
|
|
return JSONResponse(content={"status": "error", "message": "Missing 'periods' in request"}, status_code=400)
|
|
|
|
cfg = load_config()
|
|
periods_cfg = cfg.setdefault('technical_indicators', {}).setdefault('periods', {})
|
|
|
|
if indicator_name not in periods_cfg:
|
|
return JSONResponse(content={"status": "error", "message": f"Unknown indicator: {indicator_name}"}, status_code=404)
|
|
|
|
periods_cfg[indicator_name] = periods
|
|
|
|
save_config(cfg)
|
|
|
|
self.config.clear()
|
|
self.config.update(cfg)
|
|
|
|
logger.info(f"Updated {indicator_name} periods to {periods}")
|
|
return JSONResponse(content={"status": "success", "message": f"Updated {indicator_name} periods"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating indicator periods: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
# ---------------------------
|
|
# Chart and Data
|
|
# ---------------------------
|
|
|
|
@self.app.post("/api/chart-data")
|
|
async def get_chart_data(request: ChartDataRequest):
|
|
"""Get chart data for visualization"""
|
|
try:
|
|
if not self.db_manager:
|
|
logger.error("Database manager not initialized")
|
|
return JSONResponse(content={"status": "error", "message": "Database not initialized"}, status_code=500)
|
|
|
|
logger.info(f"Chart data request: symbol={request.symbol}, interval={request.interval}, limit={request.limit}")
|
|
|
|
data = await self.db_manager.get_recent_candles(request.symbol.upper(), request.interval, request.limit)
|
|
|
|
logger.info(f"Retrieved {len(data) if data else 0} candles from database")
|
|
|
|
if not data:
|
|
logger.warning(f"No data found for {request.symbol} at {request.interval}")
|
|
return JSONResponse(content={
|
|
"status": "error",
|
|
"message": f"No data found for {request.symbol} at {request.interval}. Please start data collection or download historical data first."
|
|
}, status_code=404)
|
|
|
|
logger.info(f"Returning {len(data)} candles for {request.symbol}")
|
|
return JSONResponse(content={"status": "success", "data": data})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting chart data: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": f"Error retrieving chart data: {str(e)}"}, status_code=500)
|
|
|
|
@self.app.post("/api/bulk-download")
|
|
async def bulk_download(request: BulkDownloadRequest):
|
|
"""Download historical data in bulk"""
|
|
try:
|
|
if not self.data_collector:
|
|
raise HTTPException(status_code=500, detail="Data collector not initialized")
|
|
|
|
start_date = datetime.fromisoformat(request.start_date)
|
|
end_date = datetime.fromisoformat(request.end_date) if request.end_date else datetime.utcnow()
|
|
|
|
start_date = _tz_aware(start_date)
|
|
end_date = _tz_aware(end_date)
|
|
|
|
intervals = request.intervals or ['1h', '4h', '1d']
|
|
|
|
results = []
|
|
for symbol in request.symbols:
|
|
try:
|
|
symu = symbol.upper()
|
|
|
|
# Initialize progress for UI
|
|
self.data_collector.download_progress[symu] = {
|
|
'status': 'pending',
|
|
'intervals': {i: {'status': 'pending', 'records': 0} for i in intervals},
|
|
'start_time': datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
# Spawn task
|
|
task = asyncio.create_task(
|
|
self.data_collector.bulk_download_historical_data(symu, start_date, end_date, intervals)
|
|
)
|
|
|
|
results.append({'symbol': symu, 'status': 'started', 'intervals': intervals})
|
|
logger.info(f"Bulk download started for {symbol}")
|
|
|
|
except Exception as ie:
|
|
logger.error(f"Error starting bulk download for {symbol}: {ie}")
|
|
results.append({'symbol': symu, 'status': 'error', 'error': str(ie)})
|
|
|
|
return JSONResponse(content={
|
|
"status": "success",
|
|
"message": f"Bulk download started for {len(request.symbols)} symbol(s)",
|
|
"results": results
|
|
})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error starting bulk download: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
@self.app.get("/api/download-progress")
|
|
async def get_download_progress():
|
|
"""Get progress for all active downloads"""
|
|
try:
|
|
if not self.data_collector:
|
|
return JSONResponse(content={"status": "error", "message": "Data collector not initialized"}, status_code=500)
|
|
|
|
progress = await self.data_collector.get_download_progress()
|
|
return JSONResponse(content={"status": "success", "downloads": serialize_for_json(progress)})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting download progress: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
# ---------------------------
|
|
# Environment variables
|
|
# ---------------------------
|
|
|
|
@self.app.get("/api/env")
|
|
async def get_env_vars():
|
|
"""Get environment variables"""
|
|
try:
|
|
env_vars = dotenv_values('variables.env') or {}
|
|
|
|
safe_vars = {
|
|
k: ('***' if any(s in k.upper() for s in ['SECRET', 'KEY', 'PASSWORD', 'TOKEN']) else v)
|
|
for k, v in env_vars.items()
|
|
}
|
|
|
|
return JSONResponse(content=safe_vars)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting env vars: {e}", exc_info=True)
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@self.app.post("/api/env")
|
|
async def update_env_var(env_update: EnvVarUpdate):
|
|
"""Update environment variable"""
|
|
try:
|
|
key_upper = env_update.key.upper()
|
|
display_value = env_update.value if not any(s in key_upper for s in ['PASSWORD', 'SECRET', 'KEY', 'TOKEN']) else '***'
|
|
|
|
logger.info(f"Updating env var: {env_update.key} = {display_value}")
|
|
|
|
set_key('variables.env', env_update.key, env_update.value)
|
|
reload_env_vars('variables.env')
|
|
|
|
logger.info(f"Updated and reloaded env var: {env_update.key}")
|
|
return JSONResponse(content={"status": "success", "message": f"Updated {env_update.key}"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating env var: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
@self.app.delete("/api/env/{key}")
|
|
async def delete_env_var(key: str):
|
|
"""Delete environment variable"""
|
|
try:
|
|
# Manual edit due to lack of delete in python-dotenv API
|
|
try:
|
|
with open('variables.env', 'r', encoding='utf-8') as f:
|
|
lines = f.readlines()
|
|
except FileNotFoundError:
|
|
lines = []
|
|
|
|
new_lines = [line for line in lines if not line.startswith(f"{key}=")]
|
|
|
|
with open('variables.env', 'w', encoding='utf-8') as f:
|
|
f.writelines(new_lines)
|
|
|
|
reload_env_vars('variables.env')
|
|
|
|
logger.info(f"Deleted env var: {key}")
|
|
return JSONResponse(content={"status": "success", "message": f"Deleted {key}"})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error deleting env var: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|
|
|
|
# ---------------------------
|
|
# Database stats
|
|
# ---------------------------
|
|
|
|
@self.app.get("/api/database/stats")
|
|
async def get_database_stats():
|
|
"""Get detailed database statistics"""
|
|
try:
|
|
if not self.db_manager:
|
|
raise HTTPException(status_code=500, detail="Database not initialized")
|
|
|
|
stats = await self.db_manager.get_detailed_statistics()
|
|
return JSONResponse(content={"status": "success", "stats": serialize_for_json(stats)})
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting database stats: {e}", exc_info=True)
|
|
return JSONResponse(content={"status": "error", "message": str(e)}, status_code=500)
|