k
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
import logging
|
|
from collections import deque
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, List, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 7 days of 1-minute candles
|
|
MAX_CANDLES = 7 * 24 * 60 # 10080
|
|
|
|
TIMEFRAME_MINUTES: Dict[str, int] = {
|
|
"1m": 1,
|
|
"1H": 60,
|
|
"4H": 240,
|
|
"1D": 1440,
|
|
"1W": 10080,
|
|
}
|
|
|
|
|
|
class PriceStore:
|
|
def __init__(self):
|
|
self._candles: Dict[str, deque] = {
|
|
"BTC": deque(maxlen=MAX_CANDLES),
|
|
"ETH": deque(maxlen=MAX_CANDLES),
|
|
}
|
|
|
|
def update(self, asset: str, candle: dict):
|
|
"""Add or replace the latest 1m candle for the asset."""
|
|
asset = asset.upper()
|
|
if asset not in self._candles:
|
|
self._candles[asset] = deque(maxlen=MAX_CANDLES)
|
|
buf = self._candles[asset]
|
|
# If the last candle has the same timestamp, replace it (in-progress bar)
|
|
if buf and buf[-1]["time"] == candle["time"]:
|
|
buf[-1] = candle
|
|
else:
|
|
buf.append(candle)
|
|
|
|
def get_price_at(self, asset: str, timestamp: datetime) -> Optional[float]:
|
|
"""Return the close price of the candle closest to timestamp."""
|
|
asset = asset.upper()
|
|
buf = self._candles.get(asset)
|
|
if not buf:
|
|
return None
|
|
ts = timestamp.replace(tzinfo=None)
|
|
target_unix = int(ts.timestamp()) * 1000 # candle times are ms
|
|
best = None
|
|
best_diff = float("inf")
|
|
for candle in buf:
|
|
diff = abs(candle["time"] - target_unix)
|
|
if diff < best_diff:
|
|
best_diff = diff
|
|
best = candle
|
|
return best["close"] if best else None
|
|
|
|
def get_pct_change(
|
|
self, asset: str, from_ts: datetime, minutes: int
|
|
) -> Optional[float]:
|
|
"""Return % change from from_ts to from_ts + minutes."""
|
|
asset = asset.upper()
|
|
buf = self._candles.get(asset)
|
|
if not buf:
|
|
return None
|
|
|
|
from_unix_ms = int(from_ts.replace(tzinfo=None).timestamp()) * 1000
|
|
to_unix_ms = from_unix_ms + minutes * 60 * 1000
|
|
|
|
# Find candle at from_ts
|
|
from_candle = self._closest_candle(buf, from_unix_ms)
|
|
to_candle = self._closest_candle(buf, to_unix_ms)
|
|
|
|
if from_candle is None or to_candle is None:
|
|
return None
|
|
if from_candle["close"] == 0:
|
|
return None
|
|
|
|
pct = (to_candle["close"] - from_candle["close"]) / from_candle["close"] * 100
|
|
return round(pct, 4)
|
|
|
|
def _closest_candle(self, buf: deque, target_unix_ms: int) -> Optional[dict]:
|
|
best = None
|
|
best_diff = float("inf")
|
|
for candle in buf:
|
|
diff = abs(candle["time"] - target_unix_ms)
|
|
if diff < best_diff:
|
|
best_diff = diff
|
|
best = candle
|
|
return best
|
|
|
|
def get_candles(self, asset: str, timeframe: str = "1H", limit: int = 200) -> List[dict]:
|
|
"""Aggregate 1m candles into the requested timeframe and return the last `limit` bars."""
|
|
asset = asset.upper()
|
|
buf = self._candles.get(asset)
|
|
if not buf:
|
|
return []
|
|
|
|
tf_minutes = TIMEFRAME_MINUTES.get(timeframe, 60)
|
|
|
|
if tf_minutes == 1:
|
|
candles = list(buf)[-limit:]
|
|
return [{**c, "time": c["time"] // 1000} for c in candles]
|
|
|
|
# Aggregate
|
|
aggregated: Dict[int, dict] = {}
|
|
tf_ms = tf_minutes * 60 * 1000
|
|
|
|
for candle in buf:
|
|
bucket = (candle["time"] // tf_ms) * tf_ms
|
|
if bucket not in aggregated:
|
|
aggregated[bucket] = {
|
|
"time": bucket,
|
|
"open": candle["open"],
|
|
"high": candle["high"],
|
|
"low": candle["low"],
|
|
"close": candle["close"],
|
|
"volume": candle["volume"],
|
|
}
|
|
else:
|
|
agg = aggregated[bucket]
|
|
agg["high"] = max(agg["high"], candle["high"])
|
|
agg["low"] = min(agg["low"], candle["low"])
|
|
agg["close"] = candle["close"]
|
|
agg["volume"] += candle["volume"]
|
|
|
|
sorted_candles = sorted(aggregated.values(), key=lambda c: c["time"])
|
|
result = sorted_candles[-limit:]
|
|
# Convert ms → seconds for lightweight-charts
|
|
return [{**c, "time": c["time"] // 1000} for c in result]
|
|
|
|
def latest_price(self, asset: str) -> Optional[float]:
|
|
asset = asset.upper()
|
|
buf = self._candles.get(asset)
|
|
if not buf:
|
|
return None
|
|
return buf[-1]["close"]
|
|
|
|
|
|
price_store = PriceStore()
|