k
5fb1d52026
Big-picture changes since b941223:
KOL pipeline (new) — Substack/podcast/blog RSS → AI ticker extraction →
on-chain wallet diff → talks-vs-trades divergence detection. Daily polls,
19 feeds, divergence emits Post + Telegram fan-out.
Telegram push (new) — walletless free tier + wallet-linked Pro upgrade,
in-bot preference commands (/trump /btc /funding /kol /conf /quiet),
signed-envelope API for dashboard. Disconnect-wallet keeps free
subscription.
BTC funding-rate reversal scanner (new) — hourly cron, 30d cumulative
funding threshold + mean-revert + 7d price confirm, emits via
/api/signals/ingest. BTC bottom-reversal scanner promoted to System 2.
WS broadcast rewrite — per-client send timeout + parallel fan-out
(asyncio.gather). Fixes "Binance WS no close frame" reconnect storms +
APScheduler 11-min job misses, both caused by one slow client stalling
the kline loop.
Error visibility — three silent-error sites (trumpstruth/truth_social
fetchers, funding_reversal scanner) now include exception type name so
httpx ConnectError-style empty-message errors stop logging blank lines.
Telegram bot loop now classifies ReadTimeout vs network vs unknown +
logger.exception for the unknown bucket.
Security hygiene — trumpsignal.db untracked from git (held subscriber
wallets + encrypted HL keys + 22 bot trades); .gitignore now blocks
*.db/.next/backups. CORS only allows FRONTEND_URL in production.
New ops scripts —
- scripts/preflight.py: env/DB/Telegram/AI auth verification gate
- scripts/backup_db.sh: cron-friendly daily DB backup (SQLite + Postgres)
- scripts/seed_kol_wallets.py: idempotent KOL on-chain wallet seeder
15 new Alembic migrations (007-021) covering convex strategy fields,
phase-1 safety, two-system frozen exits, invalidation prices, dynamic
SYS2 leverage, staged de-risk + pyramiding, peak gain tracking, risk
mode, auto-trade + grow flags, KOL module, KOL on-chain, KOL divergence,
Telegram bindings + walletless.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
361 lines
14 KiB
Python
361 lines
14 KiB
Python
"""
|
||
Market data abstraction — pluggable candle sources.
|
||
|
||
Currently 2 providers:
|
||
|
||
- Binance : Free public REST, broad coverage of major coins.
|
||
- Hyperliquid : SAME venue as execution. Covers HL-native perps that
|
||
Binance doesn't list (TRUMP, HYPE, PURR, etc.) and
|
||
gives us mark-price-consistent data for those assets.
|
||
|
||
Routing rule:
|
||
- Assets in HL_NATIVE_ASSETS → Hyperliquid (no Binance pair exists)
|
||
- Everything else → Binance (better history, no rate friction)
|
||
|
||
Override per-call by selecting a provider explicitly:
|
||
|
||
await BinanceCandles().fetch_4h("BTC", days=30)
|
||
await HyperliquidCandles().fetch_4h("HYPE", days=30)
|
||
|
||
# Or auto-route:
|
||
src = for_asset("HYPE") # → HyperliquidCandles
|
||
candles = await src.fetch_4h("HYPE", days=30)
|
||
|
||
Normalized candle shape (returned by ALL providers):
|
||
{"time_ms": int, "open": float, "high": float, "low": float,
|
||
"close": float, "volume": float}
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
from typing import Protocol
|
||
|
||
import httpx
|
||
|
||
from app.config import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ─── Provider protocol ──────────────────────────────────────────────────────
|
||
|
||
|
||
class CandleSource(Protocol):
|
||
name: str
|
||
|
||
async def fetch_4h(self, asset: str, days: int) -> list[dict]:
|
||
"""Last `days` worth of 4-hour candles for `asset`."""
|
||
...
|
||
|
||
async def fetch_1d(self, asset: str, days: int) -> list[dict]:
|
||
"""Last `days` daily candles. Used for SMA reclaim / VCP-Daily."""
|
||
...
|
||
|
||
async def fetch_1w(self, asset: str, weeks: int) -> list[dict]:
|
||
"""Last `weeks` weekly candles. Used for weekly-RSI reversal."""
|
||
...
|
||
|
||
async def fetch_1m(self, asset: str, start_ms: int, end_ms: int) -> list[dict]:
|
||
"""1-minute candles in [start_ms, end_ms]. Paginated internally."""
|
||
...
|
||
|
||
async def fetch_funding(self, asset: str, days: int) -> list[dict]:
|
||
"""Funding-rate history. List of {time_ms, rate}. HL-only for now —
|
||
Binance provides funding but the cross-venue rate differs, so we
|
||
defer to the execution venue (HL)."""
|
||
...
|
||
|
||
|
||
# ─── Binance ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class BinanceCandles:
|
||
name = "binance"
|
||
base_url = "https://api.binance.com/api/v3/klines"
|
||
|
||
@staticmethod
|
||
def _symbol(asset: str) -> str:
|
||
return f"{asset.upper()}USDT"
|
||
|
||
async def fetch_4h(self, asset: str, days: int) -> list[dict]:
|
||
end_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
||
start_ms = end_ms - days * 24 * 3600 * 1000
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
resp = await client.get(self.base_url, params={
|
||
"symbol": self._symbol(asset),
|
||
"interval": "4h",
|
||
"startTime": start_ms, "endTime": end_ms,
|
||
"limit": 1000,
|
||
})
|
||
resp.raise_for_status()
|
||
rows = resp.json()
|
||
return [self._normalize(r) for r in rows]
|
||
|
||
async def fetch_1d(self, asset: str, days: int) -> list[dict]:
|
||
return await self._fetch_simple(asset, "1d", days * 24 * 3600 * 1000)
|
||
|
||
async def fetch_1w(self, asset: str, weeks: int) -> list[dict]:
|
||
return await self._fetch_simple(asset, "1w", weeks * 7 * 24 * 3600 * 1000)
|
||
|
||
async def _fetch_simple(self, asset: str, interval: str, window_ms: int) -> list[dict]:
|
||
"""Single-call fetch for intervals where 1000 bars covers the window."""
|
||
end_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
||
start_ms = end_ms - window_ms
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
resp = await client.get(self.base_url, params={
|
||
"symbol": self._symbol(asset),
|
||
"interval": interval,
|
||
"startTime": start_ms, "endTime": end_ms,
|
||
"limit": 1000,
|
||
})
|
||
resp.raise_for_status()
|
||
rows = resp.json()
|
||
return [self._normalize(r) for r in rows]
|
||
|
||
async def fetch_1m(self, asset: str, start_ms: int, end_ms: int) -> list[dict]:
|
||
# Binance caps each call at 1000 candles — paginate forward.
|
||
out: list[dict] = []
|
||
cursor = start_ms
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
while cursor < end_ms:
|
||
resp = await client.get(self.base_url, params={
|
||
"symbol": self._symbol(asset),
|
||
"interval": "1m",
|
||
"startTime": cursor, "endTime": end_ms,
|
||
"limit": 1000,
|
||
})
|
||
resp.raise_for_status()
|
||
chunk = resp.json()
|
||
if not chunk:
|
||
break
|
||
out.extend(self._normalize(r) for r in chunk)
|
||
last_open = chunk[-1][0]
|
||
if last_open <= cursor:
|
||
break
|
||
cursor = last_open + 60_000
|
||
if len(chunk) < 1000:
|
||
break
|
||
return out
|
||
|
||
async def fetch_funding(self, asset: str, days: int) -> list[dict]:
|
||
"""Binance perp funding. Format: list of {time_ms, rate}.
|
||
|
||
Binance returns rate per 8h funding cycle (matches HL convention).
|
||
Note: this is Binance's perp funding, NOT HL's. For HL-traded
|
||
positions, prefer HyperliquidCandles.fetch_funding().
|
||
"""
|
||
end_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
||
start_ms = end_ms - days * 24 * 3600 * 1000
|
||
url = "https://fapi.binance.com/fapi/v1/fundingRate"
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
resp = await client.get(url, params={
|
||
"symbol": self._symbol(asset),
|
||
"startTime": start_ms, "endTime": end_ms,
|
||
"limit": 1000,
|
||
})
|
||
resp.raise_for_status()
|
||
rows = resp.json()
|
||
return [
|
||
{"time_ms": r["fundingTime"], "rate": float(r["fundingRate"])}
|
||
for r in rows
|
||
]
|
||
|
||
@staticmethod
|
||
def _normalize(row) -> dict:
|
||
return {
|
||
"time_ms": row[0],
|
||
"open": float(row[1]),
|
||
"high": float(row[2]),
|
||
"low": float(row[3]),
|
||
"close": float(row[4]),
|
||
"volume": float(row[5]),
|
||
}
|
||
|
||
|
||
# ─── Hyperliquid ────────────────────────────────────────────────────────────
|
||
|
||
|
||
class HyperliquidCandles:
|
||
"""HL public /info endpoint — same data the HL UI uses.
|
||
|
||
Endpoint:
|
||
POST https://api.hyperliquid.xyz/info
|
||
body { "type": "candleSnapshot",
|
||
"req": { "coin": "SOL", "interval": "4h",
|
||
"startTime": ms, "endTime": ms } }
|
||
|
||
Returns array of {t, T, s, i, o, c, h, l, v, n} — we normalize to our
|
||
standard shape. Useful for HL-native perps Binance doesn't list.
|
||
"""
|
||
name = "hyperliquid"
|
||
|
||
def __init__(self, mainnet: bool | None = None):
|
||
use_mainnet = settings.hl_mainnet if mainnet is None else mainnet
|
||
self.base_url = (
|
||
"https://api.hyperliquid.xyz/info" if use_mainnet
|
||
else "https://api.hyperliquid-testnet.xyz/info"
|
||
)
|
||
|
||
async def fetch_4h(self, asset: str, days: int) -> list[dict]:
|
||
return await self._fetch_window(asset, "4h", days * 24 * 3600 * 1000)
|
||
|
||
async def fetch_1d(self, asset: str, days: int) -> list[dict]:
|
||
return await self._fetch_window(asset, "1d", days * 24 * 3600 * 1000)
|
||
|
||
async def fetch_1w(self, asset: str, weeks: int) -> list[dict]:
|
||
return await self._fetch_window(asset, "1w", weeks * 7 * 24 * 3600 * 1000)
|
||
|
||
async def fetch_1m(self, asset: str, start_ms: int, end_ms: int) -> list[dict]:
|
||
# HL returns ALL candles in the window in one response — no pagination
|
||
# needed for typical scanner windows. For multi-day 1m calls HL may
|
||
# truncate; the caller should keep windows under ~24h for 1m data.
|
||
return await self._fetch(asset.upper(), "1m", start_ms, end_ms)
|
||
|
||
async def _fetch_window(self, asset: str, interval: str, window_ms: int) -> list[dict]:
|
||
end_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
||
start_ms = end_ms - window_ms
|
||
return await self._fetch(asset.upper(), interval, start_ms, end_ms)
|
||
|
||
async def _fetch(self, coin: str, interval: str, start_ms: int, end_ms: int) -> list[dict]:
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
resp = await client.post(self.base_url, json={
|
||
"type": "candleSnapshot",
|
||
"req": {
|
||
"coin": coin, "interval": interval,
|
||
"startTime": start_ms, "endTime": end_ms,
|
||
},
|
||
})
|
||
resp.raise_for_status()
|
||
rows = resp.json() or []
|
||
return [self._normalize(r) for r in rows]
|
||
|
||
async def fetch_funding(self, asset: str, days: int) -> list[dict]:
|
||
"""HL funding history — HOURLY cadence (1 cycle per hour).
|
||
|
||
IMPORTANT: HL's /info endpoint caps fundingHistory at 500 rows per
|
||
response. 500 rows × 1h cadence = 20.8 days, so a single call CAN'T
|
||
return a full 30-day window. We page backwards from `endTime` until
|
||
we cover `days` worth of history (or HL runs out of data).
|
||
|
||
Returns chronologically-sorted list of {time_ms, rate}, deduplicated.
|
||
"""
|
||
end_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
||
start_ms = end_ms - days * 24 * 3600 * 1000
|
||
cursor = end_ms
|
||
collected: dict[int, float] = {} # time_ms → rate (dedup by time)
|
||
|
||
async with httpx.AsyncClient(timeout=20) as client:
|
||
# Safety cap: at most 10 pages (5000 rows ≈ 208 days) — way more
|
||
# than any caller could reasonably want, prevents runaway loops
|
||
# if HL returns inconsistent data.
|
||
for _ in range(10):
|
||
if cursor <= start_ms:
|
||
break
|
||
resp = await client.post(self.base_url, json={
|
||
"type": "fundingHistory",
|
||
"coin": asset.upper(),
|
||
"startTime": start_ms,
|
||
"endTime": cursor,
|
||
})
|
||
resp.raise_for_status()
|
||
chunk = resp.json() or []
|
||
if not chunk:
|
||
break
|
||
for r in chunk:
|
||
t = r["time"]
|
||
if t not in collected:
|
||
collected[t] = float(r["fundingRate"])
|
||
oldest = min(r["time"] for r in chunk)
|
||
if oldest <= start_ms or len(chunk) < 500:
|
||
# Either we reached the start of our window, or HL gave
|
||
# us a partial page (no more data behind it).
|
||
break
|
||
cursor = oldest - 1
|
||
|
||
return [
|
||
{"time_ms": t, "rate": rate}
|
||
for t, rate in sorted(collected.items())
|
||
]
|
||
|
||
@staticmethod
|
||
def _normalize(row: dict) -> dict:
|
||
# HL candle keys: t=open_ms, o=open, h/l, c, v
|
||
return {
|
||
"time_ms": row["t"],
|
||
"open": float(row["o"]),
|
||
"high": float(row["h"]),
|
||
"low": float(row["l"]),
|
||
"close": float(row["c"]),
|
||
"volume": float(row["v"]),
|
||
}
|
||
|
||
|
||
# ─── Routing ────────────────────────────────────────────────────────────────
|
||
|
||
# HL-native perps that DO NOT have a Binance spot pair. The scanner / backtest
|
||
# auto-route these to HL. Add more as you discover them — the list is the
|
||
# only place to maintain provider preference.
|
||
HL_NATIVE_ASSETS = {
|
||
"HYPE", "PURR", "JEFF", "VAPOR",
|
||
"PIP", "OMNIX", "PYTH",
|
||
# NOTE: TRUMP is now listed on Binance too — using Binance for that gets
|
||
# cleaner history. SUI is on both.
|
||
}
|
||
|
||
|
||
# Reversal-strategy basket. Major-cap only (no shitcoins), 1 HL-native.
|
||
# Used by all three reversal scanners (weekly RSI, SMA reclaim, funding extreme).
|
||
REVERSAL_BASKET = ["BTC", "ETH", "SOL", "BNB", "LINK", "AAVE", "DOGE", "HYPE"]
|
||
|
||
|
||
# Bar durations in seconds — used by drop_in_progress_bar() to know whether
|
||
# the last candle in a response is still open. Crypto exchanges return the
|
||
# CURRENT (in-progress) bar in `klines` responses; for daily/weekly logic
|
||
# we usually want the most recent CLOSED bar instead.
|
||
INTERVAL_SECONDS = {
|
||
"1m": 60, "5m": 300, "15m": 900,
|
||
"1h": 3600, "4h": 14400, "8h": 28800,
|
||
"1d": 86400, "1w": 604800,
|
||
}
|
||
|
||
|
||
def drop_in_progress_bar(candles: list[dict], interval: str) -> list[dict]:
|
||
"""Return `candles` minus the last entry if it's still an open bar.
|
||
|
||
Daily / weekly bars on Binance & HL are returned with `time_ms = open_time`.
|
||
The bar's close time is open_time + interval_seconds. If now < close_time,
|
||
the bar hasn't closed yet — using its volume/close is misleading (volume
|
||
is partial, close is current spot mid-bar).
|
||
|
||
Use this in scanners that care about CONFIRMED signals (SMA reclaim,
|
||
weekly RSI). Use raw candles only when you want to react intra-bar
|
||
(rare and usually wrong for position trading).
|
||
"""
|
||
if not candles:
|
||
return candles
|
||
dur_s = INTERVAL_SECONDS.get(interval)
|
||
if dur_s is None:
|
||
return candles # unknown interval — be conservative, return as-is
|
||
bar_close_ms = candles[-1]["time_ms"] + dur_s * 1000
|
||
now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
|
||
if now_ms < bar_close_ms:
|
||
return candles[:-1]
|
||
return candles
|
||
|
||
|
||
_BINANCE_SINGLETON = BinanceCandles()
|
||
_HL_SINGLETON = HyperliquidCandles()
|
||
|
||
|
||
def for_asset(asset: str) -> CandleSource:
|
||
"""Pick the right provider. HL-native → HL, otherwise Binance.
|
||
|
||
Always returns SOMETHING — caller doesn't need to handle None. If the
|
||
asset doesn't actually trade on the chosen venue, the underlying HTTP
|
||
call will return an empty list and the caller falls back to its
|
||
"no data" path.
|
||
"""
|
||
return _HL_SINGLETON if asset.upper() in HL_NATIVE_ASSETS else _BINANCE_SINGLETON
|