Codex
146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from crypto_spot_bot.models import Candle, Ticker, utc_now
|
|
|
|
|
|
def market_quality_snapshot(
|
|
*,
|
|
symbols: list[str],
|
|
candles_by_symbol: dict[str, list[Candle]],
|
|
tickers: dict[str, Ticker],
|
|
interval: str,
|
|
) -> dict[str, Any]:
|
|
rows = [
|
|
analyze_symbol_quality(
|
|
symbol=symbol,
|
|
candles=candles_by_symbol.get(symbol, []),
|
|
ticker=tickers.get(symbol),
|
|
interval=interval,
|
|
)
|
|
for symbol in symbols
|
|
]
|
|
worst_score = min((row["score"] for row in rows), default=0.0)
|
|
status = "ok"
|
|
if any(row["status"] == "error" for row in rows):
|
|
status = "error"
|
|
elif any(row["status"] == "warn" for row in rows):
|
|
status = "warn"
|
|
return {
|
|
"status": status,
|
|
"score": round(worst_score, 4),
|
|
"symbols": rows,
|
|
}
|
|
|
|
|
|
def analyze_symbol_quality(
|
|
*,
|
|
symbol: str,
|
|
candles: list[Candle],
|
|
ticker: Ticker | None,
|
|
interval: str,
|
|
) -> dict[str, Any]:
|
|
issues: list[dict[str, Any]] = []
|
|
score = 1.0
|
|
interval_ms = _interval_ms(interval)
|
|
if not candles:
|
|
return _row(symbol, "error", 0.0, [{"code": "no_candles", "severity": "error"}], 0, None, None)
|
|
|
|
timestamps = [candle.timestamp for candle in candles]
|
|
duplicates = len(timestamps) - len(set(timestamps))
|
|
if duplicates:
|
|
issues.append({"code": "duplicate_candles", "severity": "warn", "count": duplicates})
|
|
score -= min(0.25, duplicates * 0.03)
|
|
|
|
invalid_ohlc = 0
|
|
zero_volume = 0
|
|
for candle in candles:
|
|
prices = [candle.open, candle.high, candle.low, candle.close]
|
|
if any(price <= 0 for price in prices) or candle.high < max(candle.open, candle.close) or candle.low > min(candle.open, candle.close):
|
|
invalid_ohlc += 1
|
|
if candle.volume <= 0:
|
|
zero_volume += 1
|
|
if invalid_ohlc:
|
|
issues.append({"code": "invalid_ohlc", "severity": "error", "count": invalid_ohlc})
|
|
score -= min(0.45, invalid_ohlc * 0.08)
|
|
if zero_volume:
|
|
issues.append({"code": "zero_volume", "severity": "warn", "count": zero_volume})
|
|
score -= min(0.20, zero_volume * 0.02)
|
|
|
|
missing_gaps = 0
|
|
largest_gap_ms = 0
|
|
if interval_ms > 0:
|
|
ordered = sorted(set(timestamps))
|
|
for left, right in zip(ordered, ordered[1:]):
|
|
gap = right - left
|
|
largest_gap_ms = max(largest_gap_ms, gap)
|
|
if gap > interval_ms * 1.5:
|
|
missing_gaps += max(1, round(gap / interval_ms) - 1)
|
|
if missing_gaps:
|
|
issues.append({"code": "missing_candles", "severity": "warn", "count": missing_gaps})
|
|
score -= min(0.35, missing_gaps * 0.04)
|
|
|
|
age_seconds = max(0.0, (utc_now().timestamp() * 1000 - max(timestamps)) / 1000)
|
|
stale_after = interval_ms / 1000 * 2.5
|
|
if age_seconds > stale_after:
|
|
issues.append({"code": "stale_candles", "severity": "warn", "age_seconds": round(age_seconds, 1)})
|
|
score -= 0.20
|
|
else:
|
|
age_seconds = None
|
|
|
|
if ticker is None:
|
|
issues.append({"code": "no_ticker", "severity": "error"})
|
|
score -= 0.45
|
|
else:
|
|
if ticker.last_price <= 0:
|
|
issues.append({"code": "invalid_ticker_price", "severity": "error"})
|
|
score -= 0.35
|
|
if ticker.spread_percent > 0.35:
|
|
issues.append({"code": "wide_spread", "severity": "warn", "spread_percent": round(ticker.spread_percent, 4)})
|
|
score -= 0.15
|
|
|
|
score = max(0.0, min(1.0, score))
|
|
severity = {str(issue.get("severity")) for issue in issues}
|
|
status = "error" if "error" in severity else "warn" if "warn" in severity else "ok"
|
|
return _row(
|
|
symbol,
|
|
status,
|
|
score,
|
|
issues,
|
|
len(candles),
|
|
max(timestamps),
|
|
largest_gap_ms if interval_ms > 0 else None,
|
|
)
|
|
|
|
|
|
def _row(
|
|
symbol: str,
|
|
status: str,
|
|
score: float,
|
|
issues: list[dict[str, Any]],
|
|
candle_count: int,
|
|
last_candle_timestamp: int | None,
|
|
largest_gap_ms: int | None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"symbol": symbol,
|
|
"status": status,
|
|
"score": round(score, 4),
|
|
"candle_count": candle_count,
|
|
"last_candle_timestamp": last_candle_timestamp,
|
|
"largest_gap_ms": largest_gap_ms,
|
|
"issues": issues,
|
|
}
|
|
|
|
|
|
def _interval_ms(interval: str) -> int:
|
|
normalized = str(interval).strip().upper()
|
|
if normalized == "D":
|
|
return 24 * 60 * 60 * 1000
|
|
if normalized == "W":
|
|
return 7 * 24 * 60 * 60 * 1000
|
|
if normalized.isdigit():
|
|
return int(normalized) * 60 * 1000
|
|
return 0
|