From 278d523864a4a591299c415b67bc10b9d19ccc49 Mon Sep 17 00:00:00 2001 From: Michael Feng Date: Fri, 24 Jul 2026 19:33:59 -0700 Subject: [PATCH 1/5] feat(dashboard): DEX/LP pool candles + token-symbol resolution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LP/DEX executors (connector solana-mainnet-beta, trading_pair -SOL) had no candle feed — hummingbot's CandlesFactory is CEX-only, so charts showed "Failed to load candles" — and rendered raw base-mint addresses instead of tickers. This adds, all via GeckoTerminal (the same source the pools feature and Telegram LP report already use): - Pool candles: /market/candles now detects a DEX-network connector (or an explicit pool_address) and fetches OHLCV from GeckoTerminal instead of the CEX feed. It tries the executor's own pool_address, then falls back to the base token's top live pool resolved from the mint — so a live token always charts even when the passed pool is stale/absent. Reuses handlers.dex.pool_data.fetch_ohlcv and its cache. - /market/token-symbol: server-independent mint -> ticker resolver (GeckoTerminal token info, cached 24h). - PairLabel: renders a raw base-mint pair as -, applied across the executor table + detail, agent session view (group header, chart title, positions), archived performance chart, and the agent market strip. Non-mint (CEX) pairs render unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01T9Vygff6wPFpmZb5gahuW4 --- condor/web/routes/market.py | 213 ++++++++++++++++++ .../src/components/agent/AgentMarketStrip.tsx | 9 +- .../components/agent/AgentSessionContent.tsx | 5 +- .../charts/ArchivedPerformanceCharts.tsx | 3 +- .../src/components/charts/ExecutorChart.tsx | 22 +- .../src/components/executor/ExecutorTable.tsx | 7 +- .../src/components/executor/PairLabel.tsx | 52 +++++ frontend/src/lib/api.ts | 11 + 8 files changed, 311 insertions(+), 11 deletions(-) create mode 100644 frontend/src/components/executor/PairLabel.tsx diff --git a/condor/web/routes/market.py b/condor/web/routes/market.py index bf97722c..a8a77288 100644 --- a/condor/web/routes/market.py +++ b/condor/web/routes/market.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import re import time from fastapi import APIRouter, Depends, HTTPException, Query @@ -14,6 +15,11 @@ _CANDLE_CACHE_TTL = 30.0 # seconds _CANDLE_CACHE_MAX = 50 # hard cap on entries (keys rotate every minute per chart) +# Persistent dict handed to handlers.dex.pool_data.fetch_ohlcv so its own 300s +# GeckoTerminal cache applies on top of the 30s route cache above — keeps us well +# under GeckoTerminal's free-tier rate limit when multiple pool charts are open. +_gecko_ohlcv_user_data: dict = {} + def _candle_cache_put(key: tuple, value: list, now: float) -> None: """Insert into the candle cache, evicting expired entries and capping size.""" @@ -42,6 +48,144 @@ def _candle_cache_put(key: tuple, value: list, now: float) -> None: router = APIRouter(tags=["market"]) +_MINT_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$") + + +async def _fetch_pool_candles_raw( + pool_address: str, network: str, interval: str +) -> list[CandleData]: + """OHLCV rows for one pool from GeckoTerminal (reuses handlers.dex fetch+cache). + + ``currency="token"`` prices the base token in the quote token (e.g. SOL), matching + the executor's own entry/range price scale drawn on the same chart. Returns [] on + any miss/error (never raises) so DEX pairs don't fall to the CEX 502 path. + """ + from handlers.dex.pool_data import fetch_ohlcv + + try: + ohlcv_list, err = await fetch_ohlcv( + pool_address, + network, + timeframe=interval, + currency="token", + user_data=_gecko_ohlcv_user_data, + ) + except Exception as e: + logger.warning( + "GeckoTerminal OHLCV failed pool=%s net=%s interval=%s: %s", + pool_address, + network, + interval, + e, + ) + return [] + if err or not ohlcv_list: + return [] + + candles: list[CandleData] = [] + for c in ohlcv_list: + # Rows are [timestamp, open, high, low, close, volume(_usd), (datetime)]. + if not isinstance(c, (list, tuple)) or len(c) < 6: + continue + try: + candles.append( + CandleData( + timestamp=float(c[0]), + open=float(c[1]), + high=float(c[2]), + low=float(c[3]), + close=float(c[4]), + volume=float(c[5]), + ) + ) + except (TypeError, ValueError): + continue + return candles + + +# Base-token mint → its top GeckoTerminal pool (24h-volume-sorted). Pools are stable, +# so cache for an hour. Lets an executor chart fall back to the token's live main pool +# when its own pool_address is stale/absent (e.g. a closed slot, or a multi-executor +# group where the chart picked a dead pool). +_token_pool_cache: dict[tuple[str, str], tuple[float, str]] = {} +_TOKEN_POOL_TTL = 3600.0 + + +async def _resolve_token_top_pool(mint: str, gnet: str, quote: str = "SOL") -> str: + key = (gnet, mint) + now = time.time() + cached = _token_pool_cache.get(key) + if cached and (now - cached[0]) < _TOKEN_POOL_TTL: + return cached[1] + + addr = "" + try: + import aiohttp + + url = f"https://api.geckoterminal.com/api/v2/networks/{gnet}/tokens/{mint}/pools?page=1" + async with aiohttp.ClientSession() as s: + async with s.get( + url, headers={"Accept": "application/json;version=20230302"} + ) as r: + r.raise_for_status() + data = await r.json() + pools = data.get("data") or [] + # Prefer a pool quoted in the executor's quote token (e.g. SOL) so the price + # scale matches; else the highest-volume pool (list is volume-sorted). + chosen = None + for p in pools: + parts = str((p.get("attributes") or {}).get("name") or "").upper().replace(" ", "").split("/") + if quote and quote.upper() in parts: + chosen = p + break + chosen = chosen or (pools[0] if pools else None) + if chosen: + attrs = chosen.get("attributes") or {} + addr = str(attrs.get("address") or str(chosen.get("id") or "").split("_")[-1] or "") + except Exception as e: + logger.info("top-pool resolve failed mint=%s net=%s: %s", mint, gnet, e) + addr = "" + + _token_pool_cache[key] = (now, addr) + return addr + + +async def _get_pool_candles( + connector: str, + pool_address: str | None, + trading_pair: str, + interval: str, + cache_key: tuple, + now: float, +) -> list[CandleData]: + """Candles for a DEX/LP pair from GeckoTerminal. + + Tries the executor's own ``pool_address`` first (exact pool); if that yields + nothing — a stale/closed slot, no pool_address, or a group whose first executor + sits on a dead pool — falls back to the base token's top live pool resolved from + the mint in ``trading_pair``. So a live token always charts even when the passed + pool is wrong. ``connector`` is the network id (e.g. solana-mainnet-beta). + """ + from handlers.dex.pool_data import get_gecko_network + + gnet = get_gecko_network(connector) + candles: list[CandleData] = [] + if pool_address: + candles = await _fetch_pool_candles_raw(pool_address, connector, interval) + + if not candles: + dash = trading_pair.rfind("-") + base = trading_pair[:dash] if dash > 0 else trading_pair + quote = trading_pair[dash + 1 :] if dash > 0 else "SOL" + if _MINT_RE.match(base): + top = await _resolve_token_top_pool(base, gnet, quote) + if top and top != pool_address: + candles = await _fetch_pool_candles_raw(top, connector, interval) + + _candle_cache_put(cache_key, candles, now) + return candles + + @router.get("/servers/{name}/market/connectors") async def get_connectors(name: str, user: WebUser = Depends(get_current_user)): cm = get_config_manager() @@ -242,6 +386,12 @@ async def get_candles( limit: int = Query(default=1000, ge=1, le=5000), start_time: float | None = Query(default=None, description="Unix epoch seconds"), end_time: float | None = Query(default=None, description="Unix epoch seconds"), + pool_address: str | None = Query( + default=None, + description="DEX pool address. When set, candles are fetched from " + "GeckoTerminal (by pool) instead of the CEX candle feed — used for LP/DEX " + "executors whose connector (e.g. solana-mainnet-beta) has no CandlesFactory feed.", + ), user: WebUser = Depends(get_current_user), ): cm = get_config_manager() @@ -259,12 +409,24 @@ async def get_candles( limit, bucketed_start, bucketed_end, + pool_address, ) now = time.monotonic() cached = _candle_cache.get(cache_key) if cached and (now - cached[0]) < _CANDLE_CACHE_TTL: return cached[1] + # DEX/LP pools have no CEX candle feed — route to GeckoTerminal. Trigger on a + # DEX network connector (e.g. "solana-mainnet-beta") OR an explicit pool_address, + # so these pairs never fall through to the CEX path (which 502s). _get_pool_candles + # uses the pool_address when it has data, else resolves the token's top pool. + from handlers.dex.pool_data import NETWORK_TO_GECKO + + if pool_address or connector in NETWORK_TO_GECKO: + return await _get_pool_candles( + connector, pool_address, trading_pair, interval, cache_key, now + ) + client = await cm.get_client(name) result = None try: @@ -351,3 +513,54 @@ async def get_candles( ) _candle_cache_put(cache_key, candles, now) return candles + + +# Token symbol resolution — LP/DEX executors store `trading_pair` as `-SOL` +# (Gateway can't resolve memecoins by symbol), so the dashboard shows the raw mint. +# Resolve mint → ticker via GeckoTerminal (same source as candles). Symbols are +# stable, so cache for a day. Empty string is cached too (so an unknown/illiquid +# mint doesn't re-hit GeckoTerminal every render); the UI falls back to the mint. +_token_symbol_cache: dict[tuple[str, str], tuple[float, str]] = {} +_TOKEN_SYMBOL_TTL = 24 * 3600.0 + + +@router.get("/market/token-symbol") +async def get_token_symbol( + mint: str = Query(..., description="Base token mint address"), + network: str = Query( + default="solana", description="Network id or connector (e.g. solana-mainnet-beta)" + ), + user: WebUser = Depends(get_current_user), +): + # Server-independent: pure GeckoTerminal lookup, no server scoping needed + # (auth still required). Lets the executor tables resolve symbols without + # threading a server name into every row. + from handlers.dex.pool_data import get_gecko_network + + gnet = get_gecko_network(network) + key = (gnet, mint) + now = time.time() + cached = _token_symbol_cache.get(key) + if cached and (now - cached[0]) < _TOKEN_SYMBOL_TTL: + return {"mint": mint, "symbol": cached[1]} + + symbol = "" + try: + import aiohttp + + url = f"https://api.geckoterminal.com/api/v2/networks/{gnet}/tokens/{mint}" + async with aiohttp.ClientSession() as s: + async with s.get( + url, headers={"Accept": "application/json;version=20230302"} + ) as r: + r.raise_for_status() + data = await r.json() + symbol = str( + (((data or {}).get("data") or {}).get("attributes") or {}).get("symbol") or "" + ) + except Exception as e: + logger.info("token-symbol resolve failed for mint=%s network=%s: %s", mint, gnet, e) + symbol = "" + + _token_symbol_cache[key] = (now, symbol) + return {"mint": mint, "symbol": symbol} diff --git a/frontend/src/components/agent/AgentMarketStrip.tsx b/frontend/src/components/agent/AgentMarketStrip.tsx index 126fb2b4..7849dc2f 100644 --- a/frontend/src/components/agent/AgentMarketStrip.tsx +++ b/frontend/src/components/agent/AgentMarketStrip.tsx @@ -1,5 +1,6 @@ import { useMemo } from "react"; +import { PairLabel } from "@/components/executor/PairLabel"; import { PriceTicker } from "@/components/market/PriceTicker"; import type { ExecutorInfo } from "@/lib/api"; @@ -30,9 +31,11 @@ export function AgentMarketStrip({ serverName, executors }: AgentMarketStripProp {pairs.map(({ connector, pair }) => (
- - {pair} - + {connector} diff --git a/frontend/src/components/agent/AgentSessionContent.tsx b/frontend/src/components/agent/AgentSessionContent.tsx index dfeabc1b..fca8a894 100644 --- a/frontend/src/components/agent/AgentSessionContent.tsx +++ b/frontend/src/components/agent/AgentSessionContent.tsx @@ -8,6 +8,7 @@ import { useCallback, useMemo, useState } from "react"; import { ExecutorChart, type SnapshotBubble } from "@/components/charts/ExecutorChart"; import { AgentPnlChart, metricsToDataPoints } from "@/components/agent/AgentPnlChart"; +import { PairLabel } from "@/components/executor/PairLabel"; import { useAgentExecutors } from "@/hooks/useAgentExecutors"; import { type AgentExecutorRow, type AgentPerformance, type ExecutorInfo, api } from "@/lib/api"; import { groupExecutorsByMarket } from "@/lib/executor-overlays"; @@ -365,7 +366,7 @@ export function SessionExecutors({ const current = p.current_price ?? 0; return ( - {p.trading_pair} + {side.toUpperCase()} @@ -395,7 +396,7 @@ export function SessionExecutors({ {/* Pair header (only when multiple pairs) */} {chartGroups.length > 1 && (
- {group[0].trading_pair} + {group[0].connector} = 0 ? "text-[var(--color-green)]" : "text-[var(--color-red)]"}`}> {formatCurrencyPnl(pairPnl)} diff --git a/frontend/src/components/charts/ArchivedPerformanceCharts.tsx b/frontend/src/components/charts/ArchivedPerformanceCharts.tsx index 8c6c9697..adcf63cb 100644 --- a/frontend/src/components/charts/ArchivedPerformanceCharts.tsx +++ b/frontend/src/components/charts/ArchivedPerformanceCharts.tsx @@ -1,6 +1,7 @@ import { useEffect, useMemo, useRef } from "react"; import { useQuery } from "@tanstack/react-query"; +import { PairLabel } from "@/components/executor/PairLabel"; import { api, type CandleData, type ExecutorInfo, type PnlPoint } from "@/lib/api"; import { computeMultiOverlays, @@ -586,7 +587,7 @@ export function ArchivedPerformanceCharts({ {/* Header */}

- {tradingPair} · {interval} · {executors.length} executors{overlayNote} + · {interval} · {executors.length} executors{overlayNote}

{isManyExecutors && ( diff --git a/frontend/src/components/charts/ExecutorChart.tsx b/frontend/src/components/charts/ExecutorChart.tsx index e30e0d7f..319173fd 100644 --- a/frontend/src/components/charts/ExecutorChart.tsx +++ b/frontend/src/components/charts/ExecutorChart.tsx @@ -2,6 +2,7 @@ import { useEffect, useMemo, useRef, useState, useCallback } from "react"; import { createPortal } from "react-dom"; import { useQuery } from "@tanstack/react-query"; +import { PairLabel } from "@/components/executor/PairLabel"; import { useCondorWebSocket } from "@/hooks/useWebSocket"; import { api, type ExecutorInfo } from "@/lib/api"; import { @@ -93,9 +94,24 @@ export function ExecutorChart({ const startTime = Math.floor(timeRange.start - paddingSeconds); const endTime = Math.ceil(timeRange.end + paddingSeconds); + // DEX/LP executors carry their pool address (in config or custom_info). When + // present, the backend fetches candles from GeckoTerminal by pool instead of the + // CEX candle feed the DEX connector doesn't have (which surfaced as "Failed to + // load candles"). CEX executors have no pool_address → normal candle path. + const poolAddress = useMemo(() => { + for (const ex of executors) { + const pa = + (ex.config?.pool_address as string | undefined) ?? + (ex.custom_info?.pool_address as string | undefined); + if (pa) return pa; + } + return undefined; + }, [executors]); + const { data: candles, isLoading, isError } = useQuery({ - queryKey: ["candles", server, connector, tradingPair, interval], - queryFn: () => api.getCandles(server, connector, tradingPair, interval, 5000, startTime, endTime), + queryKey: ["candles", server, connector, tradingPair, interval, poolAddress], + queryFn: () => + api.getCandles(server, connector, tradingPair, interval, 5000, startTime, endTime, poolAddress), enabled: !!server && !!connector && !!tradingPair, retry: 1, }); @@ -642,7 +658,7 @@ export function ExecutorChart({ {/* Header bar */}

- {tradingPair} · {interval} + · {interval} {hasActive && ( )} diff --git a/frontend/src/components/executor/ExecutorTable.tsx b/frontend/src/components/executor/ExecutorTable.tsx index 72986359..4dd7a9c8 100644 --- a/frontend/src/components/executor/ExecutorTable.tsx +++ b/frontend/src/components/executor/ExecutorTable.tsx @@ -11,6 +11,7 @@ import { memo, useMemo, useState } from "react"; import { useNavigate } from "react-router-dom"; import { ExecutorChart } from "@/components/charts/ExecutorChart"; +import { PairLabel } from "@/components/executor/PairLabel"; import { useResizeDrag } from "@/hooks/useResizeDrag"; import { type ExecutorInfo } from "@/lib/api"; import { @@ -174,7 +175,9 @@ const ExecutorRow = memo(function ExecutorRow({ {ex.connector} - {ex.trading_pair} + + + {executor.connector} - {executor.trading_pair} + -SOL` because Gateway can't resolve memecoins +// by symbol, so the raw mint shows in the UI unless we resolve it. +function looksLikeMint(s: string): boolean { + return /^[1-9A-HJ-NP-Za-km-z]{32,44}$/.test(s); +} + +function shortMint(s: string): string { + return s.length > 12 ? `${s.slice(0, 4)}…${s.slice(-4)}` : s; +} + +/** + * Renders a trading pair, resolving a raw base-mint to its ticker via GeckoTerminal + * (e.g. `DezXAZ…263-SOL` → `Bonk-SOL`). Non-mint pairs (normal CEX symbols) render + * unchanged. Falls back to a shortened mint while loading or if resolution fails. + */ +export function PairLabel({ + tradingPair, + connector, + className, +}: { + tradingPair: string; + connector?: string; + className?: string; +}) { + const dash = tradingPair.lastIndexOf("-"); + const base = dash > 0 ? tradingPair.slice(0, dash) : tradingPair; + const quote = dash > 0 ? tradingPair.slice(dash + 1) : ""; + const isMint = looksLikeMint(base); + + const { data } = useQuery({ + queryKey: ["token-symbol", base, connector], + queryFn: () => api.getTokenSymbol(base, connector), + enabled: isMint, + staleTime: 24 * 60 * 60 * 1000, + retry: 1, + }); + + if (!isMint) return {tradingPair}; + + const symbol = data?.symbol; + const label = symbol ? `${symbol}-${quote}` : `${shortMint(base)}-${quote}`; + return ( + + {label} + + ); +} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index b5bc6442..fede97d0 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -956,13 +956,24 @@ export const api = { limit = 1000, startTime?: number, endTime?: number, + poolAddress?: string, ) => { let url = `/api/v1/servers/${encodeURIComponent(server)}/market/candles?connector=${encodeURIComponent(connector)}&trading_pair=${encodeURIComponent(pair)}&interval=${encodeURIComponent(interval)}&limit=${limit}`; if (startTime) url += `&start_time=${startTime}`; if (endTime) url += `&end_time=${endTime}`; + // DEX/LP executors: pass the pool address so the backend fetches candles from + // GeckoTerminal (by pool) instead of the CEX candle feed the connector lacks. + if (poolAddress) url += `&pool_address=${encodeURIComponent(poolAddress)}`; return apiFetch(url); }, + // Resolve a base-token mint → ticker (GeckoTerminal). Server-independent; used to + // display LP/DEX executor pairs (stored as `-SOL`) with a readable symbol. + getTokenSymbol: (mint: string, network?: string) => + apiFetch<{ mint: string; symbol: string }>( + `/api/v1/market/token-symbol?mint=${encodeURIComponent(mint)}${network ? `&network=${encodeURIComponent(network)}` : ""}`, + ), + // ── Agents (identity + brain) ── getAgents: () => apiFetch("/api/v1/agents"), From 9010c3439ff75e6e59d2beae34a3801b3b700e92 Mon Sep 17 00:00:00 2001 From: Michael Feng Date: Fri, 24 Jul 2026 20:24:33 -0700 Subject: [PATCH 2/5] fix(dashboard): chart DEX candles against the executor's real time window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial-review follow-ups on the pool-candle path: 1. DEX candles ignored the requested window. fetch_ohlcv hardcoded limit=100 with no before_timestamp, so archived executors charted against the *latest* candles instead of the ones they traded in — overlays landed on an unrelated time axis. Thread limit + before_timestamp through fetch_ohlcv → _fetch_pool_candles_raw → _get_pool_candles, and pass the chart's window (bucketed_end) from the route. Verified against GeckoTerminal: before_timestamp walks history back to that window. limit is capped at GeckoTerminal's 1000. before_timestamp/limit are now part of fetch_ohlcv's cache key so a historical and a live window for the same pool don't collide. 2. Transient failures were cached as negative results. A single blip in _resolve_token_top_pool / get_token_symbol cached "" for 1h / 24h, blanking a resolvable pool/ticker long after the API recovered. Only cache on a successful API response (empty included — a genuinely unknown mint is worth remembering); an exception now returns uncached so the next render retries. 3. ArchivedPerformanceCharts now passes the executor's pool_address, so archived DEX charts use the exact pool the position traded in rather than the token's current top pool. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01T9Vygff6wPFpmZb5gahuW4 --- condor/web/routes/market.py | 53 +++++++++++++++---- .../charts/ArchivedPerformanceCharts.tsx | 17 +++++- handlers/dex/pool_data.py | 21 ++++++-- 3 files changed, 74 insertions(+), 17 deletions(-) diff --git a/condor/web/routes/market.py b/condor/web/routes/market.py index a8a77288..742fd770 100644 --- a/condor/web/routes/market.py +++ b/condor/web/routes/market.py @@ -52,13 +52,19 @@ def _candle_cache_put(key: tuple, value: list, now: float) -> None: async def _fetch_pool_candles_raw( - pool_address: str, network: str, interval: str + pool_address: str, + network: str, + interval: str, + limit: int = 100, + before_timestamp: int | None = None, ) -> list[CandleData]: """OHLCV rows for one pool from GeckoTerminal (reuses handlers.dex fetch+cache). ``currency="token"`` prices the base token in the quote token (e.g. SOL), matching - the executor's own entry/range price scale drawn on the same chart. Returns [] on - any miss/error (never raises) so DEX pairs don't fall to the CEX 502 path. + the executor's own entry/range price scale drawn on the same chart. ``limit`` and + ``before_timestamp`` carry the chart's requested window so an archived executor + charts against the candles it actually traded in, not the latest ones. Returns [] + on any miss/error (never raises) so DEX pairs don't fall to the CEX 502 path. """ from handlers.dex.pool_data import fetch_ohlcv @@ -69,6 +75,8 @@ async def _fetch_pool_candles_raw( timeframe=interval, currency="token", user_data=_gecko_ohlcv_user_data, + limit=limit, + before_timestamp=before_timestamp, ) except Exception as e: logger.warning( @@ -142,12 +150,14 @@ async def _resolve_token_top_pool(mint: str, gnet: str, quote: str = "SOL") -> s if chosen: attrs = chosen.get("attributes") or {} addr = str(attrs.get("address") or str(chosen.get("id") or "").split("_")[-1] or "") + # Only cache on a successful API response — addr "" here means the token + # genuinely has no pool, which is worth caching. A transient error (below) + # must not poison the cache for an hour, so it returns without caching. + _token_pool_cache[key] = (now, addr) + return addr except Exception as e: logger.info("top-pool resolve failed mint=%s net=%s: %s", mint, gnet, e) - addr = "" - - _token_pool_cache[key] = (now, addr) - return addr + return "" async def _get_pool_candles( @@ -157,6 +167,8 @@ async def _get_pool_candles( interval: str, cache_key: tuple, now: float, + limit: int = 100, + before_timestamp: int | None = None, ) -> list[CandleData]: """Candles for a DEX/LP pair from GeckoTerminal. @@ -165,13 +177,17 @@ async def _get_pool_candles( sits on a dead pool — falls back to the base token's top live pool resolved from the mint in ``trading_pair``. So a live token always charts even when the passed pool is wrong. ``connector`` is the network id (e.g. solana-mainnet-beta). + ``limit``/``before_timestamp`` carry the chart's requested window (see + :func:`_fetch_pool_candles_raw`). """ from handlers.dex.pool_data import get_gecko_network gnet = get_gecko_network(connector) candles: list[CandleData] = [] if pool_address: - candles = await _fetch_pool_candles_raw(pool_address, connector, interval) + candles = await _fetch_pool_candles_raw( + pool_address, connector, interval, limit, before_timestamp + ) if not candles: dash = trading_pair.rfind("-") @@ -180,7 +196,9 @@ async def _get_pool_candles( if _MINT_RE.match(base): top = await _resolve_token_top_pool(base, gnet, quote) if top and top != pool_address: - candles = await _fetch_pool_candles_raw(top, connector, interval) + candles = await _fetch_pool_candles_raw( + top, connector, interval, limit, before_timestamp + ) _candle_cache_put(cache_key, candles, now) return candles @@ -423,8 +441,18 @@ async def get_candles( from handlers.dex.pool_data import NETWORK_TO_GECKO if pool_address or connector in NETWORK_TO_GECKO: + # Pass the chart's window through so archived executors chart against the + # candles they actually traded in. before_timestamp = end of window (candles + # walk back from there); None = latest. GeckoTerminal caps limit at 1000. return await _get_pool_candles( - connector, pool_address, trading_pair, interval, cache_key, now + connector, + pool_address, + trading_pair, + interval, + cache_key, + now, + limit=limit, + before_timestamp=bucketed_end, ) client = await cm.get_client(name) @@ -559,8 +587,11 @@ async def get_token_symbol( (((data or {}).get("data") or {}).get("attributes") or {}).get("symbol") or "" ) except Exception as e: + # Don't cache a transient failure — a single blip must not blank this pair's + # ticker for 24h. Only successful responses (below) are cached, empty included + # (a genuinely unknown mint is worth remembering). logger.info("token-symbol resolve failed for mint=%s network=%s: %s", mint, gnet, e) - symbol = "" + return {"mint": mint, "symbol": ""} _token_symbol_cache[key] = (now, symbol) return {"mint": mint, "symbol": symbol} diff --git a/frontend/src/components/charts/ArchivedPerformanceCharts.tsx b/frontend/src/components/charts/ArchivedPerformanceCharts.tsx index adcf63cb..4138044c 100644 --- a/frontend/src/components/charts/ArchivedPerformanceCharts.tsx +++ b/frontend/src/components/charts/ArchivedPerformanceCharts.tsx @@ -256,9 +256,22 @@ export function ArchivedPerformanceCharts({ const isManyExecutors = executors.length > 15; + // DEX/LP executors carry their pool address (config or custom_info). Passing it + // lets the backend chart the exact pool this position traded in, rather than + // resolving the token's current top pool. CEX executors have none → normal path. + const poolAddress = useMemo(() => { + for (const ex of executors) { + const pa = + (ex.config?.pool_address as string | undefined) ?? + (ex.custom_info?.pool_address as string | undefined); + if (pa) return pa; + } + return undefined; + }, [executors]); + const { data: candles } = useQuery({ - queryKey: ["archived-candles", server, connector, tradingPair, fetchStart, fetchEnd, interval], - queryFn: () => api.getCandles(server, connector, tradingPair, interval, limit, fetchStart, fetchEnd), + queryKey: ["archived-candles", server, connector, tradingPair, fetchStart, fetchEnd, interval, poolAddress], + queryFn: () => api.getCandles(server, connector, tradingPair, interval, limit, fetchStart, fetchEnd, poolAddress), enabled: !!server && !!connector && !!tradingPair && timeRange.start > 0, staleTime: Infinity, retry: 1, diff --git a/handlers/dex/pool_data.py b/handlers/dex/pool_data.py index 5a3f205d..10310d39 100644 --- a/handlers/dex/pool_data.py +++ b/handlers/dex/pool_data.py @@ -123,6 +123,8 @@ async def fetch_ohlcv( timeframe: str = "1h", currency: str = "usd", user_data: dict = None, + limit: int = 100, + before_timestamp: Optional[int] = None, ) -> Tuple[Optional[List], Optional[str]]: """Fetch OHLCV data for any pool via GeckoTerminal @@ -132,6 +134,11 @@ async def fetch_ohlcv( timeframe: OHLCV timeframe ("1m", "5m", "15m", "1h", "4h", "1d") currency: Price currency - "usd" or "token" (quote token) user_data: Optional user_data dict for caching + limit: Number of candles to fetch (GeckoTerminal caps at 1000) + before_timestamp: Fetch candles ending at this unix-seconds timestamp + (walks history back from here); None = latest candles. Needed so an + archived executor charts against the price window it actually traded + in, not the latest candles. Returns: Tuple of (ohlcv_list, error_message) @@ -140,10 +147,16 @@ async def fetch_ohlcv( """ try: gecko_network = get_gecko_network(network) + # GeckoTerminal's OHLCV endpoint caps limit at 1000. + limit = max(1, min(int(limit), 1000)) - # Check cache + # Check cache. before_timestamp/limit are part of the key so a historical + # window and the live window for the same pool don't collide. if user_data is not None: - cache_key = f"ohlcv_{gecko_network}_{pool_address}_{timeframe}_{currency}" + cache_key = ( + f"ohlcv_{gecko_network}_{pool_address}_{timeframe}_{currency}" + f"_{limit}_{before_timestamp or 0}" + ) cached = get_cached(user_data, cache_key, ttl=OHLCV_CACHE_TTL) if cached is not None: return cached, None @@ -152,14 +165,14 @@ async def fetch_ohlcv( # Pass all parameters explicitly: # - currency="token" means price in quote token (not USD) # - token="base" means OHLCV for the base token - # - limit=100 for reasonable data size result = await client.get_ohlcv( gecko_network, pool_address, timeframe, + before_timestamp=before_timestamp, currency=currency, token="base", - limit=100, + limit=limit, ) # Parse response - handle different formats From 6fbe36f72f4aa9323b244c428a4ed9541e813ce7 Mon Sep 17 00:00:00 2001 From: Michael Feng Date: Fri, 24 Jul 2026 21:02:19 -0700 Subject: [PATCH 3/5] =?UTF-8?q?feat(agents):=20uniform=20bot-mode=20attrib?= =?UTF-8?q?ution=20=E2=80=94=20auto-capture=20deployed=20bot=5Fname?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bot-mode sessions whose bot name is derived at runtime (e.g. pmm_mister's {base}-mm) never persisted it, so every consumer keyed on the session's bot_name — CORE DATA totals, the dashboard session executors/PnL, the metrics timeline, the strategy performance rollup — rendered an empty session while the bot traded. Only strategies with a static name in default_config (hip_3_delta_neutral_funding_mm) attributed correctly. Root cause found on the way: the claude-agent-acp adapter sends tool arguments as ACP wire field `rawInput`, while condor read `input` — so on the claude-code path EVERY tool call arrived argument-less. That silently disabled the risk engine's deploy-cap check, dry-run action blocking, and controller_id validation, and left snapshots without Input blocks. - condor/acp/client.py: normalize rawInput -> input at the boundary, for both streamed tool_call/tool_call_update events and permission requests (fold input from updates too). PydanticAI path already sent `input`. - condor/agents/engine.py: _capture_bot_name — watch the tick's tool calls for a successful manage_bots(action="deploy") and persist the deployed bot_name into the session's config.yml (same place the delta-neutral strategy keeps its static name), preserving the file's mtime, which doubles as the session-start epoch for history window tiling. Works for every strategy with zero strategy.md changes; executor-mode strategies (lp_slot_operator) never deploy so nothing changes for them. - condor/web/routes/agents.py: session-executors endpoint now resolves the bot for non-operator (closed) sessions too, attributing that session's time-window slice of the bot's history (same tiling as _apply_bot_mode_pnl) instead of returning an empty page; response model now carries bot_name + per-controller breakdown (previously computed and dropped). - frontend/src/lib/api.ts: AgentPerformance gains optional bot_name / controllers. Known residual gap: a bot instance that vanished from the controller-performance tables AND whose archived sqlite the archived-bots analyzer can't parse (upstream "cumsum is not supported for object dtype" 500) cannot be attributed; its sessions honestly show zero. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01T9Vygff6wPFpmZb5gahuW4 --- condor/acp/client.py | 21 +++++++++-- condor/agents/engine.py | 59 ++++++++++++++++++++++++++++++ condor/web/routes/agents.py | 71 ++++++++++++++++++++++++++++++++++--- frontend/src/lib/api.ts | 3 ++ 4 files changed, 146 insertions(+), 8 deletions(-) diff --git a/condor/acp/client.py b/condor/acp/client.py index e34dfa4f..6cae4eb3 100644 --- a/condor/acp/client.py +++ b/condor/acp/client.py @@ -263,6 +263,7 @@ class ToolCallUpdate: status: str | None = None title: str | None = None output: str | None = None + input: dict | None = None @dataclass @@ -319,6 +320,8 @@ def fold_tool_call_event( tc["name"] = event.title if event.output: tc["output"] = event.output + if event.input: + tc["input"] = event.input return None @@ -679,13 +682,17 @@ def _on_session_update( if text: self._event_queue.put_nowait(ThoughtChunk(text=text)) elif kind == "tool_call": + # claude-agent-acp sends tool arguments as ``rawInput`` (ACP wire + # field), not ``input`` — without the fallback every tool call + # arrives argument-less, which silently disabled input-dependent + # consumers (risk checks, snapshots' Input blocks, bot_name capture). self._event_queue.put_nowait( ToolCallEvent( tool_call_id=update.get("toolCallId", ""), title=update.get("title", ""), status=update.get("status", "pending"), kind=update.get("kind", "other"), - input=update.get("input"), + input=update.get("input") or update.get("rawInput"), ) ) elif kind == "tool_call_update": @@ -695,6 +702,7 @@ def _on_session_update( status=update.get("status"), title=update.get("title"), output=update.get("output"), + input=update.get("input") or update.get("rawInput"), ) ) @@ -708,9 +716,16 @@ async def _on_request_permission( ) -> dict[str, Any]: options = options or [] - # If we have a permission callback, delegate to it + # If we have a permission callback, delegate to it. Normalize the ACP + # wire field ``rawInput`` into ``input`` first — the risk engine reads + # tool_call["input"], and without this every permission check saw empty + # arguments (deploy caps, dry-run blocks and controller_id validation + # all silently passed). if self.permission_callback: - return await self.permission_callback(toolCall or {}, options) + tc = dict(toolCall or {}) + if not tc.get("input") and tc.get("rawInput"): + tc["input"] = tc["rawInput"] + return await self.permission_callback(tc, options) # Default: auto-approve for opt in options: diff --git a/condor/agents/engine.py b/condor/agents/engine.py index 518eca27..237fa0ee 100644 --- a/condor/agents/engine.py +++ b/condor/agents/engine.py @@ -12,6 +12,7 @@ import asyncio import logging +import os import time from dataclasses import dataclass, field from typing import Any @@ -466,6 +467,8 @@ async def _tick(self) -> None: response_text = "".join(response_chunks) tick_duration = time.time() - self._last_tick_at + self._capture_bot_name(tool_calls) + from datetime import datetime, timezone timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") @@ -551,6 +554,62 @@ async def _collect_stream(self, acp_client: ACPClient, prompt: str): if isinstance(event, PromptDone): break + def _capture_bot_name(self, tool_calls: list[dict[str, Any]]) -> None: + """Persist the bot a session operates, observed from its own tool calls. + + Bot-mode strategies that derive the name at runtime (e.g. ``{base}-mm`` + from the traded pair) never had it in config, so every consumer keyed on + the session's ``bot_name`` — CORE DATA totals, the dashboard's per-session + executors/PnL, the metrics timeline — saw an empty session while the bot + traded. Watching ``manage_bots(action="deploy")`` here fixes that for + every strategy at once; the deployed name is written straight into the + session's ``config.yml`` (the same place the delta-neutral strategy keeps + its static name), so the merge applies from the next tick and across + restarts. Latest deploy wins — matches the framework's single + ``bot_name``-per-session model. + """ + if self.is_experiment or not self.session_dir: + return + for tc in tool_calls: + name = str(tc.get("name") or "") + if name.rsplit("__", 1)[-1] != "manage_bots": + continue + inp = tc.get("input") + if not isinstance(inp, dict) or inp.get("action") != "deploy": + continue + # A deploy the risk engine blocked (or that errored) never created a + # bot — don't attribute the session to a name that isn't running. + if str(tc.get("status") or "") == "failed": + continue + bot_name = str(inp.get("bot_name") or "").strip() + if not bot_name or bot_name == self.config.get("bot_name"): + continue + from .config import save_full_config + + self.config["bot_name"] = bot_name + try: + # config.yml's mtime doubles as the session start epoch for + # bot-history window tiling (_session_start_epoch) — preserve it + # across this mid-session rewrite or the session's PnL window + # would silently shift to the deploy tick. + cfg_path = self.session_dir / "config.yml" + stat = cfg_path.stat() if cfg_path.exists() else None + save_full_config(self.session_dir, self.config) + if stat is not None: + os.utime(cfg_path, (stat.st_atime, stat.st_mtime)) + except Exception: + log.exception( + "TickEngine %s: failed to persist bot_name=%s", + self.agent_id, + bot_name, + ) + continue + log.info( + "TickEngine %s: captured deployed bot_name=%s into session config", + self.agent_id, + bot_name, + ) + # ------------------------------------------------------------------ # Client factory # ------------------------------------------------------------------ diff --git a/condor/web/routes/agents.py b/condor/web/routes/agents.py index 60ff38f2..575076b8 100644 --- a/condor/web/routes/agents.py +++ b/condor/web/routes/agents.py @@ -157,6 +157,11 @@ class AgentPerformanceModel(BaseModel): open_count: int = 0 closed_count: int = 0 executors: list[dict[str, Any]] = [] + # Bot-mode attribution: which bot this session operates (resolved instance + # name) and its per-controller breakdown, so the UI can label bot-mode + # sessions and filter live executors by the bot's controller ids. + bot_name: str = "" + controllers: list[dict[str, Any]] = [] class StrategyPerformanceResponse(BaseModel): @@ -380,6 +385,54 @@ def _session_start_epoch(strategy_dir: Path, num: int) -> float: return 0.0 +async def _merge_session_bot_slice( + client: Any, + perf: Any, + bot_base: str, + strategy_dir: Path, + session_num: int, + session_nums: list[int], +) -> None: + """Fold one closed session's slice of its bot's history into ``perf`` in place. + + Same window tiling as :func:`_apply_bot_mode_pnl` — ``[start_N, start_next)`` + where ``start_next`` is the next session's start (or now for the last one) — + so the per-session detail agrees with the strategy performance rollup. + Realized-only: a closed session never carries the bot's live unrealized PnL + or open positions, which belong to the current operator. + """ + from condor.fetchers.bot_performance import ( + fetch_all_bot_performance, + fetch_instance_history, + resolve_bot_instances, + slice_history, + ) + + start = _session_start_epoch(strategy_dir, session_num) + if start <= 0: + return + later = [n for n in session_nums if n > session_num] + end = _session_start_epoch(strategy_dir, min(later)) if later else time.time() + if end <= start: + end = time.time() + + try: + all_bot_perf = await fetch_all_bot_performance(client) + except Exception as e: + log.warning("session bot slice: fetch_all_bot_performance failed: %s", e) + return + instances = resolve_bot_instances(all_bot_perf, bot_base) + if not instances: + return + histories = [await fetch_instance_history(client, inst) for inst in instances] + realized, volume, trades = slice_history(histories, start, end) + perf.realized_pnl += realized + perf.total_pnl = perf.realized_pnl + perf.unrealized_pnl + perf.volume += volume + perf.trade_count += int(round(trades)) + perf.bot_name = perf.bot_name or bot_base + + async def _apply_bot_mode_pnl( real_sessions: list, strategy_dir: Path, default_config: dict | None, client: Any ) -> None: @@ -1174,12 +1227,18 @@ async def get_session_executors( if k == "session" ] is_operator = bool(session_nums) and session_num == max(session_nums) - bot_name = ( - _session_bot_base(strategy.dir, strategy.default_config, session_num) - if is_operator - else "" + bot_base = _session_bot_base(strategy.dir, strategy.default_config, session_num) + perf = await fetch_agent_performance( + client, agent_id, bot_name=bot_base if is_operator else "" ) - perf = await fetch_agent_performance(client, agent_id, bot_name=bot_name) + if bot_base and not is_operator: + # Closed bot-mode session: attribute its own time-window slice of the + # bot's history (same tiling as _apply_bot_mode_pnl) so a finished run + # still shows what it earned instead of an empty page. No live rows — + # open positions belong to the current operator. + await _merge_session_bot_slice( + client, perf, bot_base, strategy.dir, session_num, session_nums + ) model = AgentPerformanceModel( agent_id=agent_id, session_num=session_num, @@ -1193,6 +1252,8 @@ async def get_session_executors( open_count=perf.open_count, closed_count=perf.closed_count, executors=perf.executors, + bot_name=perf.bot_name, + controllers=perf.controllers, ) return {"executors": perf.executors, "performance": model.model_dump()} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index fede97d0..5efede20 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -408,6 +408,9 @@ export interface AgentPerformance { open_count: number; closed_count: number; executors: AgentExecutorRow[]; + // Bot-mode sessions: resolved bot instance name + per-controller breakdown + bot_name?: string; + controllers?: Record[]; } export interface AgentPerformanceResponse { From 421df77a6273883004b11aa32ae70c8457cb3dd0 Mon Sep 17 00:00:00 2001 From: Michael Feng Date: Sat, 25 Jul 2026 06:35:09 -0700 Subject: [PATCH 4/5] fix(dashboard): evict stale OHLCV cache entries + validate mint/pool_address params MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The OHLCV cache key now includes limit/before_timestamp, which rotates by the minute for live charts — with no eviction on the get/set path the web route's process-lifetime dict grew without bound (~0.3MB per entry). Sweep expired entries on every cache miss, same contract cached_call already has. Also reject a mint (base58) or pool_address (base58/0x-hex) that doesn't look like an address before interpolating it into a GeckoTerminal URL. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01T9Vygff6wPFpmZb5gahuW4 --- _shared.py | 676 ++++++++++++++++++++++++++++++++++++ condor/web/routes/market.py | 14 + handlers/dex/_shared.py | 5 + handlers/dex/pool_data.py | 7 +- market.py | 611 ++++++++++++++++++++++++++++++++ pool_data.py | 465 +++++++++++++++++++++++++ 6 files changed, 1777 insertions(+), 1 deletion(-) create mode 100644 _shared.py create mode 100644 market.py create mode 100644 pool_data.py diff --git a/_shared.py b/_shared.py new file mode 100644 index 00000000..21098270 --- /dev/null +++ b/_shared.py @@ -0,0 +1,676 @@ +""" +Shared utilities for DEX trading handlers + +Contains: +- Server client helper +- Explorer URL generation +- Common formatters +- Conversation-level caching (delegates to condor.cache) +""" + +import logging +from typing import Any, Callable, Dict, List, Optional + +from condor.cache import ( + DEFAULT_CACHE_TTL, + clear_cache as _clear_cache, + evict_expired as _evict_expired, + get_cached as _get_cached, + invalidate_groups as _invalidate_groups, + invalidates as _invalidates, + set_cached as _set_cached, + cached_call as _cached_call, +) + +logger = logging.getLogger(__name__) + + +# ============================================ +# CONVERSATION-LEVEL CACHE (thin wrappers) +# ============================================ + +_NS = "_cache" # namespace for DEX cache + + +def get_cached(user_data: dict, key: str, ttl: int = DEFAULT_CACHE_TTL) -> Optional[Any]: + return _get_cached(user_data, key, ttl, namespace=_NS) + + +def set_cached(user_data: dict, key: str, value: Any) -> None: + _set_cached(user_data, key, value, namespace=_NS) + + +def evict_expired(user_data: dict) -> int: + return _evict_expired(user_data, namespace=_NS) + + +def clear_cache(user_data: dict, key: Optional[str] = None) -> None: + _clear_cache(user_data, key, namespace=_NS) + + +async def cached_call( + user_data: dict, + key: str, + fetch_func: Callable, + ttl: int = DEFAULT_CACHE_TTL, + *args, + **kwargs, +) -> Any: + return await _cached_call(user_data, key, fetch_func, ttl, *args, namespace=_NS, **kwargs) + + +# ============================================ +# CACHE INVALIDATION GROUPS +# ============================================ + +CACHE_GROUPS = { + "balances": [ + "gateway_balances", + "portfolio_data", + "wallet_balances", + "token_balances", + "gateway_data", + ], + "positions": [ + "clmm_positions", + "liquidity_positions", + "pool_positions", + "gateway_lp_positions", + "gateway_closed_positions", + ], + "swaps": ["swap_history", "recent_swaps"], + "tokens": ["token_cache"], + "all": None, +} + + +def invalidate_cache(user_data: dict, *groups: str) -> None: + """Invalidate cache keys by group name(s).""" + # Handle special direct-on-user_data keys for backward compat + for group in groups: + if group == "all": + user_data.pop("token_cache", None) + else: + keys = CACHE_GROUPS.get(group, [group]) + if keys: + for key in keys: + if key in user_data: + user_data.pop(key, None) + _invalidate_groups(user_data, CACHE_GROUPS, *groups, namespace=_NS) + + +def invalidates(*groups: str): + """Decorator that invalidates cache groups after handler execution.""" + return _invalidates(*groups, groups_map=CACHE_GROUPS, namespace=_NS) + + +# ============================================ +# SERVER CLIENT HELPERS +# ============================================ + +from config_manager import get_client + +# ============================================ +# EXPLORER URL GENERATION +# ============================================ + +SOLANA_EXPLORERS = { + "orb": "https://orb.helius.dev/tx/{tx_hash}?cluster={cluster}&tab=summary", + "solscan": "https://solscan.io/tx/{tx_hash}", + "solana_explorer": "https://explorer.solana.com/tx/{tx_hash}", +} + +ETHEREUM_EXPLORERS = { + "etherscan": "https://etherscan.io/tx/{tx_hash}", + "arbiscan": "https://arbiscan.io/tx/{tx_hash}", + "basescan": "https://basescan.org/tx/{tx_hash}", +} + + +def get_explorer_url(tx_hash: str, network: str) -> Optional[str]: + """Generate explorer URL for a transaction + + Args: + tx_hash: Transaction hash/signature + network: Network name (e.g., 'solana-mainnet-beta', 'ethereum-mainnet') + + Returns: + Explorer URL or None if network not supported + """ + if not tx_hash: + return None + + if network.startswith("solana"): + # Use Orb explorer for Solana (Helius) + cluster = "mainnet-beta" if "mainnet" in network else "devnet" + return SOLANA_EXPLORERS["orb"].format(tx_hash=tx_hash, cluster=cluster) + elif "ethereum" in network or "mainnet" in network: + if "arbitrum" in network: + return ETHEREUM_EXPLORERS["arbiscan"].format(tx_hash=tx_hash) + elif "base" in network: + return ETHEREUM_EXPLORERS["basescan"].format(tx_hash=tx_hash) + else: + return ETHEREUM_EXPLORERS["etherscan"].format(tx_hash=tx_hash) + + return None + + +def get_explorer_name(network: str) -> str: + """Get the explorer name for display + + Args: + network: Network name + + Returns: + Explorer name (e.g., 'Orb', 'Etherscan') + """ + if network.startswith("solana"): + return "Orb" + elif "arbitrum" in network: + return "Arbiscan" + elif "base" in network: + return "Basescan" + elif "ethereum" in network: + return "Etherscan" + return "Explorer" + + +# ============================================ +# SWAP FORMATTERS +# ============================================ + + +def format_swap_summary(swap: Dict[str, Any], include_explorer: bool = True) -> str: + """Format a swap record for display + + Args: + swap: Swap data dictionary + include_explorer: Whether to include explorer link + + Returns: + Formatted swap summary string (not escaped) + """ + pair = swap.get("trading_pair", "N/A") + side = swap.get("side", "N/A") + status = swap.get("status", "N/A") + network = swap.get("network", "") + tx_hash = swap.get("transaction_hash", "") + + # Format amounts + input_amount = swap.get("input_amount") + output_amount = swap.get("output_amount") + base_token = swap.get("base_token", "") + quote_token = swap.get("quote_token", "") + + # Build amount string + if input_amount is not None and output_amount is not None: + if side == "BUY": + # Buying base with quote + amount_str = f"{_format_amount(output_amount)} {base_token} for {_format_amount(input_amount)} {quote_token}" + else: + # Selling base for quote + amount_str = f"{_format_amount(input_amount)} {base_token} for {_format_amount(output_amount)} {quote_token}" + elif input_amount is not None: + amount_str = f"{_format_amount(input_amount)}" + else: + amount_str = "N/A" + + # Format price + price = swap.get("price") + price_str = f"@ {_format_price(price)}" if price else "" + + # Build the line + parts = [f"{side} {pair}", amount_str] + if price_str: + parts.append(price_str) + parts.append(f"[{status}]") + + return " ".join(parts) + + +def format_swap_detail(swap: Dict[str, Any]) -> str: + """Format detailed swap information + + Args: + swap: Swap data dictionary + + Returns: + Formatted multi-line swap details (not escaped) + """ + lines = [] + + # Header with status emoji + status = swap.get("status", "UNKNOWN") + status_emoji = get_status_emoji(status) + lines.append(f"{status_emoji} Swap Details") + lines.append("") + + # Trading info + pair = swap.get("trading_pair", "N/A") + side = swap.get("side", "N/A") + lines.append(f"Pair: {pair}") + lines.append(f"Side: {side}") + + # Amounts + input_amount = swap.get("input_amount") + output_amount = swap.get("output_amount") + base_token = swap.get("base_token", "") + quote_token = swap.get("quote_token", "") + + if input_amount is not None: + lines.append( + f"Input: {_format_amount(input_amount)} {quote_token if side == 'BUY' else base_token}" + ) + if output_amount is not None: + lines.append( + f"Output: {_format_amount(output_amount)} {base_token if side == 'BUY' else quote_token}" + ) + + # Price + price = swap.get("price") + if price: + lines.append(f"Price: {_format_price(price)}") + + # Slippage + slippage = swap.get("slippage_pct") + if slippage is not None: + lines.append(f"Slippage: {slippage}%") + + # Network info + lines.append("") + connector = swap.get("connector", "N/A") + network = swap.get("network", "N/A") + lines.append(f"Connector: {connector}") + lines.append(f"Network: {network}") + + # Transaction + tx_hash = swap.get("transaction_hash", "") + if tx_hash: + lines.append(f"Tx: {tx_hash[:16]}...") + + # Timestamp + timestamp = swap.get("timestamp", "") + if timestamp: + # Format timestamp for display + if "T" in timestamp: + date_part = timestamp.split("T")[0] + time_part = ( + timestamp.split("T")[1].split(".")[0] + if "." in timestamp.split("T")[1] + else timestamp.split("T")[1].split("+")[0] + ) + lines.append(f"Time: {date_part} {time_part}") + + # Status + lines.append(f"Status: {status}") + + return "\n".join(lines) + + +def get_status_emoji(status: str) -> str: + """Get emoji for swap status + + Args: + status: Status string (CONFIRMED, PENDING, FAILED, etc.) + + Returns: + Emoji character + """ + status_emojis = { + "CONFIRMED": "✅", + "PENDING": "⏳", + "FAILED": "❌", + "REJECTED": "🚫", + "UNKNOWN": "❓", + } + return status_emojis.get(status.upper(), "📊") + + +def _format_amount(amount: float) -> str: + """Format amount with appropriate precision""" + if amount is None: + return "N/A" + + if amount == 0: + return "0" + + # Use appropriate decimal places based on size + if abs(amount) >= 1000: + return f"{amount:,.2f}" + elif abs(amount) >= 1: + return f"{amount:.4f}" + elif abs(amount) >= 0.0001: + return f"{amount:.6f}" + else: + return f"{amount:.8f}" + + +def _format_price(price: float) -> str: + """Format price with appropriate precision""" + if price is None: + return "N/A" + + if price == 0: + return "0" + + if abs(price) >= 1: + return f"{price:.4f}" + elif abs(price) >= 0.0001: + return f"{price:.6f}" + else: + return f"{price:.10f}" + + +# ============================================ +# RELATIVE TIME FORMATTER +# ============================================ + + +def format_relative_time(timestamp: str) -> str: + """Format timestamp as relative time (e.g., '53s', '22m', '1h', '2d') + + Args: + timestamp: ISO format timestamp string + + Returns: + Relative time string + """ + from datetime import datetime, timezone + + if not timestamp: + return "" + + try: + # Parse ISO timestamp + if "T" in timestamp: + # Handle various ISO formats + ts_str = timestamp.replace("Z", "+00:00") + if "." in ts_str: + # Remove microseconds if present + parts = ts_str.split(".") + if "+" in parts[1]: + ts_str = parts[0] + "+" + parts[1].split("+")[1] + elif "-" in parts[1]: + ts_str = parts[0] + "-" + parts[1].split("-", 1)[1] + else: + ts_str = parts[0] + + # Parse with timezone + try: + dt = datetime.fromisoformat(ts_str) + except ValueError: + # Fallback: try without timezone + dt = datetime.fromisoformat(timestamp.split("+")[0].split(".")[0]) + dt = dt.replace(tzinfo=timezone.utc) + else: + return "" + + # Calculate difference + now = datetime.now(timezone.utc) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + + diff = now - dt + seconds = int(diff.total_seconds()) + + if seconds < 0: + return "now" + elif seconds < 60: + return f"{seconds}s" + elif seconds < 3600: + return f"{seconds // 60}m" + elif seconds < 86400: + return f"{seconds // 3600}h" + else: + return f"{seconds // 86400}d" + + except Exception as e: + logger.debug(f"Error formatting relative time: {e}") + return "" + + +# ============================================ +# STATE HELPERS +# ============================================ + + +def clear_dex_state(context) -> None: + """Clear all DEX-related state from user context + + Args: + context: Telegram context object + """ + context.user_data.pop("dex_state", None) + context.user_data.pop("dex_previous_state", None) + context.user_data.pop("quote_swap_params", None) + context.user_data.pop("execute_swap_params", None) + + +# ============================================ +# HISTORY FILTER & PAGINATION HELPERS +# ============================================ + +from dataclasses import dataclass +from typing import Literal + +HistoryType = Literal["swap", "position"] + +# Available filter options per history type +HISTORY_FILTERS = { + "swap": { + "trading_pair": ["All", "SOL-USDC", "SOL-ORE", "ORE-USDC", "ETH-USDC"], + "connector": ["All", "jupiter", "uniswap"], + "status": ["All", "CONFIRMED", "PENDING", "FAILED"], + }, + "position": { + "trading_pair": ["All", "SOL-USDC", "ORE-SOL", "METv-SOL"], + "connector": ["All", "meteora", "orca", "raydium"], + "status": ["All", "OPEN", "CLOSED"], + }, +} + +DEFAULT_PAGE_SIZE = 10 + + +@dataclass +class HistoryFilters: + """Stores filter and pagination state for history views""" + + history_type: HistoryType = "swap" + trading_pair: Optional[str] = None # None = All + connector: Optional[str] = None # None = All + status: Optional[str] = None # None = All + network: Optional[str] = None # None = All + offset: int = 0 + limit: int = DEFAULT_PAGE_SIZE + total_count: int = 0 + + def to_dict(self) -> Dict[str, Any]: + return { + "history_type": self.history_type, + "trading_pair": self.trading_pair, + "connector": self.connector, + "status": self.status, + "network": self.network, + "offset": self.offset, + "limit": self.limit, + "total_count": self.total_count, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "HistoryFilters": + return cls( + history_type=data.get("history_type", "swap"), + trading_pair=data.get("trading_pair"), + connector=data.get("connector"), + status=data.get("status"), + network=data.get("network"), + offset=data.get("offset", 0), + limit=data.get("limit", DEFAULT_PAGE_SIZE), + total_count=data.get("total_count", 0), + ) + + def reset_pagination(self) -> None: + """Reset pagination when filters change""" + self.offset = 0 + + @property + def current_page(self) -> int: + return (self.offset // self.limit) + 1 + + @property + def total_pages(self) -> int: + if self.total_count == 0: + return 1 + return (self.total_count + self.limit - 1) // self.limit + + @property + def has_next(self) -> bool: + return self.offset + self.limit < self.total_count + + @property + def has_prev(self) -> bool: + return self.offset > 0 + + +def get_history_filters(user_data: dict, history_type: HistoryType) -> HistoryFilters: + """Get current history filters from user data""" + key = f"history_filters_{history_type}" + data = user_data.get(key) + if data: + return HistoryFilters.from_dict(data) + return HistoryFilters(history_type=history_type) + + +def set_history_filters(user_data: dict, filters: HistoryFilters) -> None: + """Save history filters to user data""" + key = f"history_filters_{filters.history_type}" + user_data[key] = filters.to_dict() + + +def build_filter_buttons( + filters: HistoryFilters, callback_prefix: str +) -> List[List["InlineKeyboardButton"]]: + """Build filter button rows for history views + + Args: + filters: Current filter state + callback_prefix: Prefix for callback data (e.g., "dex:swap_hist" or "dex:lp_hist") + + Returns: + List of button rows + """ + from telegram import InlineKeyboardButton + + rows = [] + + # Trading pair filter + pair_label = filters.trading_pair or "All Pairs" + rows.append( + [ + InlineKeyboardButton( + f"💱 {pair_label}", callback_data=f"{callback_prefix}_filter_pair" + ), + ] + ) + + # Connector & Status filters (same row) + connector_label = filters.connector or "All DEX" + status_label = filters.status or "All Status" + rows.append( + [ + InlineKeyboardButton( + f"🔌 {connector_label}", + callback_data=f"{callback_prefix}_filter_connector", + ), + InlineKeyboardButton( + f"📊 {status_label}", callback_data=f"{callback_prefix}_filter_status" + ), + ] + ) + + return rows + + +def build_pagination_buttons( + filters: HistoryFilters, callback_prefix: str +) -> List["InlineKeyboardButton"]: + """Build pagination buttons for history views + + Args: + filters: Current filter state with pagination info + callback_prefix: Prefix for callback data + + Returns: + List of buttons for a single row + """ + from telegram import InlineKeyboardButton + + buttons = [] + + # Previous button + if filters.has_prev: + buttons.append( + InlineKeyboardButton("« Prev", callback_data=f"{callback_prefix}_page_prev") + ) + else: + buttons.append(InlineKeyboardButton(" ", callback_data="dex:noop")) + + # Page indicator + page_text = f"{filters.current_page}/{filters.total_pages}" + buttons.append(InlineKeyboardButton(page_text, callback_data="dex:noop")) + + # Next button + if filters.has_next: + buttons.append( + InlineKeyboardButton("Next »", callback_data=f"{callback_prefix}_page_next") + ) + else: + buttons.append(InlineKeyboardButton(" ", callback_data="dex:noop")) + + return buttons + + +def build_filter_selection_keyboard( + options: List[str], + current_value: Optional[str], + callback_prefix: str, + back_callback: str, +) -> "InlineKeyboardMarkup": + """Build a keyboard for selecting a filter value + + Args: + options: List of available options + current_value: Currently selected value (None = All) + callback_prefix: Prefix for callback data + back_callback: Callback for back button + + Returns: + InlineKeyboardMarkup with option buttons + """ + from telegram import InlineKeyboardButton, InlineKeyboardMarkup + + buttons = [] + row = [] + + for opt in options: + # Check if this option is currently selected + is_selected = (opt == "All" and current_value is None) or (opt == current_value) + label = f"✓ {opt}" if is_selected else opt + + # Use None for "All" option + value = "" if opt == "All" else opt + row.append( + InlineKeyboardButton(label, callback_data=f"{callback_prefix}_{value}") + ) + + if len(row) == 2: + buttons.append(row) + row = [] + + if row: + buttons.append(row) + + buttons.append([InlineKeyboardButton("« Back", callback_data=back_callback)]) + + return InlineKeyboardMarkup(buttons) diff --git a/condor/web/routes/market.py b/condor/web/routes/market.py index 742fd770..d6c8c732 100644 --- a/condor/web/routes/market.py +++ b/condor/web/routes/market.py @@ -50,6 +50,10 @@ def _candle_cache_put(key: tuple, value: list, now: float) -> None: _MINT_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$") +# DEX pool addresses across GeckoTerminal networks: base58 (Solana) or 0x-hex +# (EVM). Used to sanitize the pool_address query param before it reaches a URL. +_POOL_ADDR_RE = re.compile(r"^[A-Za-z0-9]{16,90}$") + async def _fetch_pool_candles_raw( pool_address: str, @@ -416,6 +420,11 @@ async def get_candles( if not cm.has_server_access(user.id, name): raise HTTPException(status_code=403, detail="No access") + # pool_address is interpolated into GeckoTerminal URLs — restrict to plain + # address characters (base58 for Solana, 0x-hex for EVM networks). + if pool_address and not _POOL_ADDR_RE.match(pool_address): + raise HTTPException(status_code=400, detail="Invalid pool_address") + # Bucket start_time to 60s intervals so near-identical requests share cache bucketed_start = int(start_time // 60) * 60 if start_time is not None else None bucketed_end = int(end_time // 60) * 60 if end_time is not None else None @@ -565,6 +574,11 @@ async def get_token_symbol( # threading a server name into every row. from handlers.dex.pool_data import get_gecko_network + # The mint is interpolated into the GeckoTerminal URL path — reject anything + # that isn't a base58 pubkey (mirrors the frontend's looksLikeMint gate). + if not _MINT_RE.match(mint): + raise HTTPException(status_code=400, detail="Invalid mint address") + gnet = get_gecko_network(network) key = (gnet, mint) now = time.time() diff --git a/handlers/dex/_shared.py b/handlers/dex/_shared.py index ab903e9b..21098270 100644 --- a/handlers/dex/_shared.py +++ b/handlers/dex/_shared.py @@ -14,6 +14,7 @@ from condor.cache import ( DEFAULT_CACHE_TTL, clear_cache as _clear_cache, + evict_expired as _evict_expired, get_cached as _get_cached, invalidate_groups as _invalidate_groups, invalidates as _invalidates, @@ -39,6 +40,10 @@ def set_cached(user_data: dict, key: str, value: Any) -> None: _set_cached(user_data, key, value, namespace=_NS) +def evict_expired(user_data: dict) -> int: + return _evict_expired(user_data, namespace=_NS) + + def clear_cache(user_data: dict, key: Optional[str] = None) -> None: _clear_cache(user_data, key, namespace=_NS) diff --git a/handlers/dex/pool_data.py b/handlers/dex/pool_data.py index 10310d39..815cf5dd 100644 --- a/handlers/dex/pool_data.py +++ b/handlers/dex/pool_data.py @@ -14,7 +14,7 @@ from config_manager import get_client -from ._shared import get_cached, set_cached +from ._shared import evict_expired, get_cached, set_cached logger = logging.getLogger(__name__) @@ -160,6 +160,11 @@ async def fetch_ohlcv( cached = get_cached(user_data, cache_key, ttl=OHLCV_CACHE_TTL) if cached is not None: return cached, None + # Sweep stale entries on every miss (same contract as cached_call). + # Keys now include limit/before_timestamp, which rotates by the minute + # for live charts — without eviction a long-lived process (the web + # route's persistent dict) grows without bound. + evict_expired(user_data) client = GeckoTerminalAsyncClient() # Pass all parameters explicitly: diff --git a/market.py b/market.py new file mode 100644 index 00000000..d6c8c732 --- /dev/null +++ b/market.py @@ -0,0 +1,611 @@ +from __future__ import annotations + +import logging +import re +import time + +from fastapi import APIRouter, Depends, HTTPException, Query + +from config_manager import get_config_manager + +logger = logging.getLogger(__name__) + +# Simple TTL cache for candle data +_candle_cache: dict[tuple, tuple[float, list]] = {} # key -> (timestamp, data) +_CANDLE_CACHE_TTL = 30.0 # seconds +_CANDLE_CACHE_MAX = 50 # hard cap on entries (keys rotate every minute per chart) + +# Persistent dict handed to handlers.dex.pool_data.fetch_ohlcv so its own 300s +# GeckoTerminal cache applies on top of the 30s route cache above — keeps us well +# under GeckoTerminal's free-tier rate limit when multiple pool charts are open. +_gecko_ohlcv_user_data: dict = {} + + +def _candle_cache_put(key: tuple, value: list, now: float) -> None: + """Insert into the candle cache, evicting expired entries and capping size.""" + expired = [ + k for k, (ts, _) in _candle_cache.items() if now - ts >= _CANDLE_CACHE_TTL + ] + for k in expired: + _candle_cache.pop(k, None) + _candle_cache[key] = (now, value) + while len(_candle_cache) > _CANDLE_CACHE_MAX: + # dicts preserve insertion order: drop the oldest entry first + _candle_cache.pop(next(iter(_candle_cache))) + + +from condor.web.auth import get_current_user +from condor.web.models import ( + CandleData, + MarketPriceResponse, + OrderBookLevel, + OrderBookResponse, + TradingRuleItem, + TradingRulesResponse, + WebUser, +) + +router = APIRouter(tags=["market"]) + + +_MINT_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$") + +# DEX pool addresses across GeckoTerminal networks: base58 (Solana) or 0x-hex +# (EVM). Used to sanitize the pool_address query param before it reaches a URL. +_POOL_ADDR_RE = re.compile(r"^[A-Za-z0-9]{16,90}$") + + +async def _fetch_pool_candles_raw( + pool_address: str, + network: str, + interval: str, + limit: int = 100, + before_timestamp: int | None = None, +) -> list[CandleData]: + """OHLCV rows for one pool from GeckoTerminal (reuses handlers.dex fetch+cache). + + ``currency="token"`` prices the base token in the quote token (e.g. SOL), matching + the executor's own entry/range price scale drawn on the same chart. ``limit`` and + ``before_timestamp`` carry the chart's requested window so an archived executor + charts against the candles it actually traded in, not the latest ones. Returns [] + on any miss/error (never raises) so DEX pairs don't fall to the CEX 502 path. + """ + from handlers.dex.pool_data import fetch_ohlcv + + try: + ohlcv_list, err = await fetch_ohlcv( + pool_address, + network, + timeframe=interval, + currency="token", + user_data=_gecko_ohlcv_user_data, + limit=limit, + before_timestamp=before_timestamp, + ) + except Exception as e: + logger.warning( + "GeckoTerminal OHLCV failed pool=%s net=%s interval=%s: %s", + pool_address, + network, + interval, + e, + ) + return [] + if err or not ohlcv_list: + return [] + + candles: list[CandleData] = [] + for c in ohlcv_list: + # Rows are [timestamp, open, high, low, close, volume(_usd), (datetime)]. + if not isinstance(c, (list, tuple)) or len(c) < 6: + continue + try: + candles.append( + CandleData( + timestamp=float(c[0]), + open=float(c[1]), + high=float(c[2]), + low=float(c[3]), + close=float(c[4]), + volume=float(c[5]), + ) + ) + except (TypeError, ValueError): + continue + return candles + + +# Base-token mint → its top GeckoTerminal pool (24h-volume-sorted). Pools are stable, +# so cache for an hour. Lets an executor chart fall back to the token's live main pool +# when its own pool_address is stale/absent (e.g. a closed slot, or a multi-executor +# group where the chart picked a dead pool). +_token_pool_cache: dict[tuple[str, str], tuple[float, str]] = {} +_TOKEN_POOL_TTL = 3600.0 + + +async def _resolve_token_top_pool(mint: str, gnet: str, quote: str = "SOL") -> str: + key = (gnet, mint) + now = time.time() + cached = _token_pool_cache.get(key) + if cached and (now - cached[0]) < _TOKEN_POOL_TTL: + return cached[1] + + addr = "" + try: + import aiohttp + + url = f"https://api.geckoterminal.com/api/v2/networks/{gnet}/tokens/{mint}/pools?page=1" + async with aiohttp.ClientSession() as s: + async with s.get( + url, headers={"Accept": "application/json;version=20230302"} + ) as r: + r.raise_for_status() + data = await r.json() + pools = data.get("data") or [] + # Prefer a pool quoted in the executor's quote token (e.g. SOL) so the price + # scale matches; else the highest-volume pool (list is volume-sorted). + chosen = None + for p in pools: + parts = str((p.get("attributes") or {}).get("name") or "").upper().replace(" ", "").split("/") + if quote and quote.upper() in parts: + chosen = p + break + chosen = chosen or (pools[0] if pools else None) + if chosen: + attrs = chosen.get("attributes") or {} + addr = str(attrs.get("address") or str(chosen.get("id") or "").split("_")[-1] or "") + # Only cache on a successful API response — addr "" here means the token + # genuinely has no pool, which is worth caching. A transient error (below) + # must not poison the cache for an hour, so it returns without caching. + _token_pool_cache[key] = (now, addr) + return addr + except Exception as e: + logger.info("top-pool resolve failed mint=%s net=%s: %s", mint, gnet, e) + return "" + + +async def _get_pool_candles( + connector: str, + pool_address: str | None, + trading_pair: str, + interval: str, + cache_key: tuple, + now: float, + limit: int = 100, + before_timestamp: int | None = None, +) -> list[CandleData]: + """Candles for a DEX/LP pair from GeckoTerminal. + + Tries the executor's own ``pool_address`` first (exact pool); if that yields + nothing — a stale/closed slot, no pool_address, or a group whose first executor + sits on a dead pool — falls back to the base token's top live pool resolved from + the mint in ``trading_pair``. So a live token always charts even when the passed + pool is wrong. ``connector`` is the network id (e.g. solana-mainnet-beta). + ``limit``/``before_timestamp`` carry the chart's requested window (see + :func:`_fetch_pool_candles_raw`). + """ + from handlers.dex.pool_data import get_gecko_network + + gnet = get_gecko_network(connector) + candles: list[CandleData] = [] + if pool_address: + candles = await _fetch_pool_candles_raw( + pool_address, connector, interval, limit, before_timestamp + ) + + if not candles: + dash = trading_pair.rfind("-") + base = trading_pair[:dash] if dash > 0 else trading_pair + quote = trading_pair[dash + 1 :] if dash > 0 else "SOL" + if _MINT_RE.match(base): + top = await _resolve_token_top_pool(base, gnet, quote) + if top and top != pool_address: + candles = await _fetch_pool_candles_raw( + top, connector, interval, limit, before_timestamp + ) + + _candle_cache_put(cache_key, candles, now) + return candles + + +@router.get("/servers/{name}/market/connectors") +async def get_connectors(name: str, user: WebUser = Depends(get_current_user)): + cm = get_config_manager() + if not cm.has_server_access(user.id, name): + raise HTTPException(status_code=403, detail="No access") + + from condor.server_data_service import ServerDataType, get_server_data_service + + try: + result = await get_server_data_service().get_or_fetch( + name, ServerDataType.CANDLE_CONNECTORS + ) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) + return result + + +@router.get("/servers/{name}/market/connected-exchanges") +async def get_connected_exchanges(name: str, user: WebUser = Depends(get_current_user)): + """Get connectors that have credentials configured (accounts connected).""" + cm = get_config_manager() + if not cm.has_server_access(user.id, name): + raise HTTPException(status_code=403, detail="No access") + + from condor.server_data_service import ServerDataType, get_server_data_service + + try: + result = await get_server_data_service().get_or_fetch( + name, ServerDataType.CONNECTORS + ) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) + return result or [] + + +@router.get("/servers/{name}/market/prices", response_model=MarketPriceResponse) +async def get_price( + name: str, + connector: str = Query(...), + trading_pair: str = Query(...), + user: WebUser = Depends(get_current_user), +): + cm = get_config_manager() + if not cm.has_server_access(user.id, name): + raise HTTPException(status_code=403, detail="No access") + + from condor.server_data_service import ServerDataType, get_server_data_service + + try: + result = await get_server_data_service().get_or_fetch( + name, + ServerDataType.PRICES, + connector_name=connector, + trading_pair=trading_pair, + ) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) + + if result is None: + raise HTTPException(status_code=502, detail="Failed to fetch price") + + if isinstance(result, (int, float)): + return MarketPriceResponse( + connector=connector, trading_pair=trading_pair, mid_price=float(result) + ) + elif isinstance(result, dict): + return MarketPriceResponse( + connector=connector, + trading_pair=trading_pair, + mid_price=float(result.get("mid_price", result.get("price", 0))), + best_bid=float(result.get("best_bid", 0)), + best_ask=float(result.get("best_ask", 0)), + ) + raise HTTPException(status_code=502, detail="Unexpected response format") + + +@router.post("/servers/{name}/rate-oracle/rates") +async def get_rate_oracle_rates( + name: str, + body: dict, + user: WebUser = Depends(get_current_user), +): + cm = get_config_manager() + if not cm.has_server_access(user.id, name): + raise HTTPException(status_code=403, detail="No access") + + trading_pairs = body.get("trading_pairs", []) + if not trading_pairs: + return {"rates": {}} + + client = await cm.get_client(name) + try: + result = await client.rate_oracle.get_rates(trading_pairs=trading_pairs) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) + + return result + + +@router.get("/servers/{name}/market/trading-rules", response_model=TradingRulesResponse) +async def get_trading_rules( + name: str, + connector: str = Query(...), + user: WebUser = Depends(get_current_user), +): + cm = get_config_manager() + if not cm.has_server_access(user.id, name): + raise HTTPException(status_code=403, detail="No access") + + from condor.server_data_service import ServerDataType, get_server_data_service + + try: + result = await get_server_data_service().get_or_fetch( + name, ServerDataType.TRADING_RULES, connector_name=connector + ) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) + + if not isinstance(result, dict): + return TradingRulesResponse(connector=connector, rules=[]) + + rules = [] + for pair, rule_data in result.items(): + if isinstance(rule_data, dict): + rules.append( + TradingRuleItem( + trading_pair=pair, + min_order_size=float(rule_data.get("min_order_size", 0)), + min_notional_size=float(rule_data.get("min_notional_size", 0)), + min_price_increment=float(rule_data.get("min_price_increment", 0)), + min_base_amount_increment=float( + rule_data.get("min_base_amount_increment", 0) + ), + ) + ) + return TradingRulesResponse(connector=connector, rules=rules) + + +@router.get("/servers/{name}/market/order-book", response_model=OrderBookResponse) +async def get_order_book( + name: str, + connector: str = Query(...), + trading_pair: str = Query(...), + depth: int = Query(default=20, ge=1, le=100), + user: WebUser = Depends(get_current_user), +): + cm = get_config_manager() + if not cm.has_server_access(user.id, name): + raise HTTPException(status_code=403, detail="No access") + + client = await cm.get_client(name) + try: + result = await client.market_data.get_order_book( + connector_name=connector, trading_pair=trading_pair + ) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) + + bids = [] + asks = [] + if isinstance(result, dict): + for entry in (result.get("bids") or [])[:depth]: + if isinstance(entry, (list, tuple)) and len(entry) >= 2: + bids.append( + OrderBookLevel(price=float(entry[0]), amount=float(entry[1])) + ) + elif isinstance(entry, dict): + bids.append( + OrderBookLevel( + price=float(entry.get("price", 0)), + amount=float(entry.get("amount", entry.get("quantity", 0))), + ) + ) + for entry in (result.get("asks") or [])[:depth]: + if isinstance(entry, (list, tuple)) and len(entry) >= 2: + asks.append( + OrderBookLevel(price=float(entry[0]), amount=float(entry[1])) + ) + elif isinstance(entry, dict): + asks.append( + OrderBookLevel( + price=float(entry.get("price", 0)), + amount=float(entry.get("amount", entry.get("quantity", 0))), + ) + ) + + return OrderBookResponse( + connector=connector, trading_pair=trading_pair, bids=bids, asks=asks + ) + + +@router.get("/servers/{name}/market/candles", response_model=list[CandleData]) +async def get_candles( + name: str, + connector: str = Query(...), + trading_pair: str = Query(...), + interval: str = Query(default="1m"), + limit: int = Query(default=1000, ge=1, le=5000), + start_time: float | None = Query(default=None, description="Unix epoch seconds"), + end_time: float | None = Query(default=None, description="Unix epoch seconds"), + pool_address: str | None = Query( + default=None, + description="DEX pool address. When set, candles are fetched from " + "GeckoTerminal (by pool) instead of the CEX candle feed — used for LP/DEX " + "executors whose connector (e.g. solana-mainnet-beta) has no CandlesFactory feed.", + ), + user: WebUser = Depends(get_current_user), +): + cm = get_config_manager() + if not cm.has_server_access(user.id, name): + raise HTTPException(status_code=403, detail="No access") + + # pool_address is interpolated into GeckoTerminal URLs — restrict to plain + # address characters (base58 for Solana, 0x-hex for EVM networks). + if pool_address and not _POOL_ADDR_RE.match(pool_address): + raise HTTPException(status_code=400, detail="Invalid pool_address") + + # Bucket start_time to 60s intervals so near-identical requests share cache + bucketed_start = int(start_time // 60) * 60 if start_time is not None else None + bucketed_end = int(end_time // 60) * 60 if end_time is not None else None + cache_key = ( + name, + connector, + trading_pair, + interval, + limit, + bucketed_start, + bucketed_end, + pool_address, + ) + now = time.monotonic() + cached = _candle_cache.get(cache_key) + if cached and (now - cached[0]) < _CANDLE_CACHE_TTL: + return cached[1] + + # DEX/LP pools have no CEX candle feed — route to GeckoTerminal. Trigger on a + # DEX network connector (e.g. "solana-mainnet-beta") OR an explicit pool_address, + # so these pairs never fall through to the CEX path (which 502s). _get_pool_candles + # uses the pool_address when it has data, else resolves the token's top pool. + from handlers.dex.pool_data import NETWORK_TO_GECKO + + if pool_address or connector in NETWORK_TO_GECKO: + # Pass the chart's window through so archived executors chart against the + # candles they actually traded in. before_timestamp = end of window (candles + # walk back from there); None = latest. GeckoTerminal caps limit at 1000. + return await _get_pool_candles( + connector, + pool_address, + trading_pair, + interval, + cache_key, + now, + limit=limit, + before_timestamp=bucketed_end, + ) + + client = await cm.get_client(name) + result = None + try: + # Prefer historical candles with time range when start_time is given + if start_time is not None: + st = int(start_time) + et = int(end_time) if end_time else int(time.time()) + logger.info( + "Fetching historical candles: connector=%s pair=%s interval=%s start=%s end=%s", + connector, + trading_pair, + interval, + st, + et, + ) + result = await client.market_data.get_historical_candles( + connector, + trading_pair, + interval, + start_time=st, + end_time=et, + ) + logger.info( + "Historical candles result: type=%s len=%s", + type(result).__name__, + len(result) if isinstance(result, (list, dict)) else "?", + ) + except Exception as e: + logger.warning( + "get_historical_candles failed: %s — falling back to get_candles", e + ) + result = None + + # Fallback: if historical returned nothing usable, use regular candles + candles_raw = ( + result + if isinstance(result, list) + else result.get("data", []) if isinstance(result, dict) else [] + ) + if not candles_raw: + try: + logger.info( + "Falling back to get_candles: connector=%s pair=%s interval=%s limit=%s", + connector, + trading_pair, + interval, + limit, + ) + result = await client.market_data.get_candles( + connector, trading_pair, interval, limit + ) + except Exception as e: + raise HTTPException(status_code=502, detail=str(e)) + + candles_raw = ( + result + if isinstance(result, list) + else result.get("data", []) if isinstance(result, dict) else [] + ) + + candles = [] + for c in candles_raw: + if isinstance(c, dict): + candles.append( + CandleData( + timestamp=float(c.get("timestamp", 0)), + open=float(c.get("open", 0)), + high=float(c.get("high", 0)), + low=float(c.get("low", 0)), + close=float(c.get("close", 0)), + volume=float(c.get("volume", 0)), + ) + ) + elif isinstance(c, (list, tuple)) and len(c) >= 6: + candles.append( + CandleData( + timestamp=float(c[0]), + open=float(c[1]), + high=float(c[2]), + low=float(c[3]), + close=float(c[4]), + volume=float(c[5]), + ) + ) + _candle_cache_put(cache_key, candles, now) + return candles + + +# Token symbol resolution — LP/DEX executors store `trading_pair` as `-SOL` +# (Gateway can't resolve memecoins by symbol), so the dashboard shows the raw mint. +# Resolve mint → ticker via GeckoTerminal (same source as candles). Symbols are +# stable, so cache for a day. Empty string is cached too (so an unknown/illiquid +# mint doesn't re-hit GeckoTerminal every render); the UI falls back to the mint. +_token_symbol_cache: dict[tuple[str, str], tuple[float, str]] = {} +_TOKEN_SYMBOL_TTL = 24 * 3600.0 + + +@router.get("/market/token-symbol") +async def get_token_symbol( + mint: str = Query(..., description="Base token mint address"), + network: str = Query( + default="solana", description="Network id or connector (e.g. solana-mainnet-beta)" + ), + user: WebUser = Depends(get_current_user), +): + # Server-independent: pure GeckoTerminal lookup, no server scoping needed + # (auth still required). Lets the executor tables resolve symbols without + # threading a server name into every row. + from handlers.dex.pool_data import get_gecko_network + + # The mint is interpolated into the GeckoTerminal URL path — reject anything + # that isn't a base58 pubkey (mirrors the frontend's looksLikeMint gate). + if not _MINT_RE.match(mint): + raise HTTPException(status_code=400, detail="Invalid mint address") + + gnet = get_gecko_network(network) + key = (gnet, mint) + now = time.time() + cached = _token_symbol_cache.get(key) + if cached and (now - cached[0]) < _TOKEN_SYMBOL_TTL: + return {"mint": mint, "symbol": cached[1]} + + symbol = "" + try: + import aiohttp + + url = f"https://api.geckoterminal.com/api/v2/networks/{gnet}/tokens/{mint}" + async with aiohttp.ClientSession() as s: + async with s.get( + url, headers={"Accept": "application/json;version=20230302"} + ) as r: + r.raise_for_status() + data = await r.json() + symbol = str( + (((data or {}).get("data") or {}).get("attributes") or {}).get("symbol") or "" + ) + except Exception as e: + # Don't cache a transient failure — a single blip must not blank this pair's + # ticker for 24h. Only successful responses (below) are cached, empty included + # (a genuinely unknown mint is worth remembering). + logger.info("token-symbol resolve failed for mint=%s network=%s: %s", mint, gnet, e) + return {"mint": mint, "symbol": ""} + + _token_symbol_cache[key] = (now, symbol) + return {"mint": mint, "symbol": symbol} diff --git a/pool_data.py b/pool_data.py new file mode 100644 index 00000000..815cf5dd --- /dev/null +++ b/pool_data.py @@ -0,0 +1,465 @@ +""" +Pool Data Utilities + +Provides unified data fetching for DEX pools: +- OHLCV data via GeckoTerminal (works for any pool on any DEX) +- Liquidity/bin data via Gateway CLMM (for supported DEXes) +- Pool info normalization across different sources +""" + +import logging +from typing import Any, Dict, List, Optional, Tuple + +from geckoterminal_py import GeckoTerminalAsyncClient + +from config_manager import get_client + +from ._shared import evict_expired, get_cached, set_cached + +logger = logging.getLogger(__name__) + +# Supported DEXes for liquidity data (via gateway CLMM) +LIQUIDITY_SUPPORTED_DEXES = { + "meteora": "solana", + "raydium": "solana", + "orca": "solana", + "uniswap": "ethereum", + "pancakeswap": "bsc", +} + +# GeckoTerminal network mapping +NETWORK_TO_GECKO = { + "solana": "solana", + "solana-mainnet-beta": "solana", + "ethereum": "eth", + "ethereum-mainnet": "eth", + "arbitrum": "arbitrum", + "arbitrum-one": "arbitrum", + "base": "base", + "base-mainnet": "base", + "bsc": "bsc", + "binance-smart-chain": "bsc", + "polygon": "polygon_pos", + "polygon-mainnet": "polygon_pos", + "avalanche": "avalanche", + "optimism": "optimism", +} + +# DEX ID to GeckoTerminal DEX mapping +DEX_TO_GECKO = { + "meteora": "meteora", + "raydium": "raydium", + "orca": "orca", + "uniswap": "uniswap", + "uniswap_v3": "uniswap_v3", + "pancakeswap": "pancakeswap", + "pancakeswap_v3": "pancakeswap_v3", + "sushiswap": "sushiswap", +} + +# Cache TTLs +OHLCV_CACHE_TTL = 300 # 5 minutes +BINS_CACHE_TTL = 60 # 1 minute + + +def get_gecko_network(network: str) -> str: + """Convert internal network name to GeckoTerminal network ID""" + return NETWORK_TO_GECKO.get(network, network) + + +def can_fetch_liquidity(dex_id: str, network: str = None) -> bool: + """Check if liquidity/bin data can be fetched for this DEX + + Args: + dex_id: DEX identifier (e.g., "meteora", "raydium") + network: Optional network to verify (must be Solana for now) + + Returns: + True if liquidity data is available via gateway CLMM + """ + dex_lower = dex_id.lower() if dex_id else "" + + if dex_lower not in LIQUIDITY_SUPPORTED_DEXES: + return False + + if network: + expected_network = LIQUIDITY_SUPPORTED_DEXES.get(dex_lower) + gecko_network = get_gecko_network(network) + if gecko_network != expected_network: + return False + + return True + + +def get_connector_for_dex(dex_id: str) -> Optional[str]: + """Get the gateway connector name for a DEX ID + + Args: + dex_id: DEX identifier from GeckoTerminal + + Returns: + Connector name for gateway CLMM or None + """ + dex_lower = dex_id.lower() if dex_id else "" + + # Direct mapping + if dex_lower in LIQUIDITY_SUPPORTED_DEXES: + return dex_lower + + # Handle variations + if "meteora" in dex_lower: + return "meteora" + if "raydium" in dex_lower: + return "raydium" + if "orca" in dex_lower: + return "orca" + + return None + + +async def fetch_ohlcv( + pool_address: str, + network: str, + timeframe: str = "1h", + currency: str = "usd", + user_data: dict = None, + limit: int = 100, + before_timestamp: Optional[int] = None, +) -> Tuple[Optional[List], Optional[str]]: + """Fetch OHLCV data for any pool via GeckoTerminal + + Args: + pool_address: Pool contract address + network: Network identifier (will be converted to GeckoTerminal format) + timeframe: OHLCV timeframe ("1m", "5m", "15m", "1h", "4h", "1d") + currency: Price currency - "usd" or "token" (quote token) + user_data: Optional user_data dict for caching + limit: Number of candles to fetch (GeckoTerminal caps at 1000) + before_timestamp: Fetch candles ending at this unix-seconds timestamp + (walks history back from here); None = latest candles. Needed so an + archived executor charts against the price window it actually traded + in, not the latest candles. + + Returns: + Tuple of (ohlcv_list, error_message) + ohlcv_list: List of [timestamp, open, high, low, close, volume] or None + error_message: Error string if failed, None on success + """ + try: + gecko_network = get_gecko_network(network) + # GeckoTerminal's OHLCV endpoint caps limit at 1000. + limit = max(1, min(int(limit), 1000)) + + # Check cache. before_timestamp/limit are part of the key so a historical + # window and the live window for the same pool don't collide. + if user_data is not None: + cache_key = ( + f"ohlcv_{gecko_network}_{pool_address}_{timeframe}_{currency}" + f"_{limit}_{before_timestamp or 0}" + ) + cached = get_cached(user_data, cache_key, ttl=OHLCV_CACHE_TTL) + if cached is not None: + return cached, None + # Sweep stale entries on every miss (same contract as cached_call). + # Keys now include limit/before_timestamp, which rotates by the minute + # for live charts — without eviction a long-lived process (the web + # route's persistent dict) grows without bound. + evict_expired(user_data) + + client = GeckoTerminalAsyncClient() + # Pass all parameters explicitly: + # - currency="token" means price in quote token (not USD) + # - token="base" means OHLCV for the base token + result = await client.get_ohlcv( + gecko_network, + pool_address, + timeframe, + before_timestamp=before_timestamp, + currency=currency, + token="base", + limit=limit, + ) + + # Parse response - handle different formats + ohlcv_list = None + + try: + import pandas as pd + + if isinstance(result, pd.DataFrame): + if not result.empty: + # Convert DataFrame to list format + ohlcv_list = result.values.tolist() + except ImportError: + pass + + if ohlcv_list is None: + if isinstance(result, list): + ohlcv_list = result + elif isinstance(result, dict): + # Try nested structure + data = result.get("data", result) + if isinstance(data, dict): + attrs = data.get("attributes", data) + ohlcv_list = attrs.get("ohlcv_list", []) + elif isinstance(data, list): + ohlcv_list = data + + if not ohlcv_list: + return None, "No OHLCV data available" + + # Debug logging: show price range from OHLCV data + if ohlcv_list: + try: + closes = [float(c[4]) for c in ohlcv_list if len(c) > 4 and c[4]] + if closes: + logger.info( + f"OHLCV {pool_address[:8]}... {timeframe} currency={currency}: " + f"{len(ohlcv_list)} candles, price range [{min(closes):.6f} - {max(closes):.6f}]" + ) + except Exception as e: + logger.debug(f"Could not log OHLCV price range: {e}") + + # Cache result + if user_data is not None: + set_cached(user_data, cache_key, ohlcv_list) + + return ohlcv_list, None + + except Exception as e: + logger.error(f"Error fetching OHLCV: {e}", exc_info=True) + return None, f"Failed to fetch OHLCV: {str(e)}" + + +async def fetch_liquidity_bins( + pool_address: str, + connector: str = "meteora", + network: str = "solana-mainnet-beta", + user_data: dict = None, + chat_id: int = None, + context=None, +) -> Tuple[Optional[List], Optional[Dict], Optional[str]]: + """Fetch liquidity bin data for CLMM pools via gateway + + Args: + pool_address: Pool contract address + connector: DEX connector (meteora, raydium, orca) + network: Network identifier + user_data: Optional user_data dict for caching + chat_id: Chat ID for per-chat server selection + + Returns: + Tuple of (bins_list, pool_info, error_message) + bins_list: List of bin dicts with price, base_token_amount, quote_token_amount + pool_info: Full pool info dict + error_message: Error string if failed, None on success + """ + try: + if not can_fetch_liquidity(connector): + return None, None, f"Liquidity data not available for {connector}" + + # Check cache + cache_key = f"pool_bins_{connector}_{pool_address}" + if user_data is not None: + cached = get_cached(user_data, cache_key, ttl=BINS_CACHE_TTL) + if cached is not None: + return cached.get("bins"), cached, None + + client = await get_client(chat_id, context=context) + if not client: + return None, None, "Gateway client not available" + + pool_info = None + + # First try get_pool_info (works for pools known to gateway) + try: + pool_info = await client.gateway_clmm.get_pool_info( + connector=connector, network=network, pool_address=pool_address + ) + except Exception as e: + # If get_pool_info fails (e.g., pool not in gateway config or not a DLMM pool), + # try finding the pool via get_pools search + error_str = str(e) + if "validation error" in error_str.lower() or "Field required" in error_str: + logger.info( + f"Pool {pool_address[:12]}... not found via get_pool_info, trying get_pools search" + ) + try: + # Search for pool by address using get_pools + search_result = await client.gateway_clmm.get_pools( + connector=connector, search_term=pool_address, limit=1 + ) + pools = search_result.get("pools", []) + if pools: + # Found the pool, but get_pools doesn't include bins + # Return pool info without bins - caller can handle this + pool_info = pools[0] + pool_info["address"] = pool_address + logger.info( + f"Found pool via get_pools: {pool_info.get('trading_pair', 'Unknown')}" + ) + else: + # Pool not found in DLMM pools - might be an AMM pool or non-existent + logger.info( + f"Pool {pool_address[:12]}... not found in {connector} DLMM pools" + ) + return ( + None, + None, + f"Pool not found in {connector} DLMM pools. This may be an AMM pool or not a {connector} pool.", + ) + except Exception as search_e: + logger.warning(f"get_pools search also failed: {search_e}") + return ( + None, + None, + f"Could not fetch pool info. Pool may not be a {connector} DLMM pool.", + ) + + if pool_info is None: + # Re-raise with a cleaner message for non-validation errors + return None, None, f"Failed to fetch pool: {str(e)[:100]}" + + if not pool_info: + return None, None, "Pool not found" + + bins = pool_info.get("bins", []) + + # Cache result + if user_data is not None: + set_cached(user_data, cache_key, pool_info) + + return bins, pool_info, None + + except Exception as e: + logger.error(f"Error fetching liquidity bins: {e}", exc_info=True) + return None, None, f"Failed to fetch liquidity: {str(e)}" + + +def normalize_pool_data(pool: dict, source: str = "gecko") -> Dict[str, Any]: + """Normalize pool data from different sources to a common format + + Args: + pool: Raw pool data dict + source: Data source ("gecko" or "gateway") + + Returns: + Normalized pool dict with consistent keys + """ + if source == "gecko": + # GeckoTerminal format + attrs = pool.get("attributes", pool) + + return { + "address": attrs.get("address") or pool.get("id", "").split("_")[-1], + "name": attrs.get("name", "Unknown"), + "base_token_symbol": attrs.get("base_token_symbol", "???"), + "quote_token_symbol": attrs.get("quote_token_symbol", "???"), + "base_token_price_usd": attrs.get("base_token_price_usd"), + "quote_token_price_usd": attrs.get("quote_token_price_usd"), + "network": pool.get("network") or attrs.get("network", "solana"), + "dex_id": attrs.get("dex_id", "unknown"), + "reserve_usd": attrs.get("reserve_in_usd"), + "volume_24h": _get_nested_float(attrs, "volume_usd", "h24"), + "volume_6h": _get_nested_float(attrs, "volume_usd", "h6"), + "volume_1h": _get_nested_float(attrs, "volume_usd", "h1"), + "price_change_24h": _get_nested_float( + attrs, "price_change_percentage", "h24" + ), + "price_change_6h": _get_nested_float( + attrs, "price_change_percentage", "h6" + ), + "price_change_1h": _get_nested_float( + attrs, "price_change_percentage", "h1" + ), + "fdv_usd": attrs.get("fdv_usd"), + "market_cap_usd": attrs.get("market_cap_usd"), + "pool_created_at": attrs.get("pool_created_at"), + "source": "gecko", + } + + elif source == "gateway": + # Gateway CLMM format + return { + "address": pool.get("pool_address") or pool.get("address", ""), + "name": pool.get("trading_pair") or pool.get("name", "Unknown"), + "base_token_symbol": pool.get("base_symbol", "???"), + "quote_token_symbol": pool.get("quote_symbol", "???"), + "base_token_price_usd": None, # Not provided by gateway + "quote_token_price_usd": None, + "network": "solana", + "dex_id": pool.get("connector", "meteora"), + "reserve_usd": pool.get("liquidity") or pool.get("tvl"), + "volume_24h": pool.get("volume_24h"), + "price_change_24h": None, + "current_price": pool.get("current_price") or pool.get("price"), + "bin_step": pool.get("bin_step"), + "apr": pool.get("apr"), + "apy": pool.get("apy"), + "base_fee_percentage": pool.get("base_fee_percentage"), + "mint_x": pool.get("mint_x"), + "mint_y": pool.get("mint_y"), + "source": "gateway", + } + + return pool + + +def _get_nested_float(data: dict, *keys) -> Optional[float]: + """Get a nested float value from dict, trying multiple key patterns""" + # Try nested access + value = data + for key in keys: + if isinstance(value, dict): + value = value.get(key) + else: + value = None + break + + if value is not None: + try: + return float(value) + except (ValueError, TypeError): + pass + + # Try flattened key with underscore + flat_key = "_".join(keys) + value = data.get(flat_key) + if value is not None: + try: + return float(value) + except (ValueError, TypeError): + pass + + # Try flattened key with dot + flat_key = ".".join(keys) + value = data.get(flat_key) + if value is not None: + try: + return float(value) + except (ValueError, TypeError): + pass + + return None + + +def extract_pair_from_name(name: str) -> Tuple[str, str]: + """Extract base and quote symbols from pool name + + Args: + name: Pool name like "SOL/USDC" or "SOL-USDC" or "SOL / USDC" + + Returns: + Tuple of (base_symbol, quote_symbol) + """ + if not name: + return "???", "???" + + # Try different separators + for sep in ["/", " / ", "-", " - "]: + if sep in name: + parts = name.split(sep) + if len(parts) >= 2: + return parts[0].strip(), parts[1].strip() + + return name, "???" From b9759acbe9f952aaeb130cee7159caf89078bc3e Mon Sep 17 00:00:00 2001 From: Michael Feng Date: Sat, 25 Jul 2026 06:35:54 -0700 Subject: [PATCH 5/5] chore: remove stray files accidentally copied to repo root Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01T9Vygff6wPFpmZb5gahuW4 --- _shared.py | 676 --------------------------------------------------- market.py | 611 ---------------------------------------------- pool_data.py | 465 ----------------------------------- 3 files changed, 1752 deletions(-) delete mode 100644 _shared.py delete mode 100644 market.py delete mode 100644 pool_data.py diff --git a/_shared.py b/_shared.py deleted file mode 100644 index 21098270..00000000 --- a/_shared.py +++ /dev/null @@ -1,676 +0,0 @@ -""" -Shared utilities for DEX trading handlers - -Contains: -- Server client helper -- Explorer URL generation -- Common formatters -- Conversation-level caching (delegates to condor.cache) -""" - -import logging -from typing import Any, Callable, Dict, List, Optional - -from condor.cache import ( - DEFAULT_CACHE_TTL, - clear_cache as _clear_cache, - evict_expired as _evict_expired, - get_cached as _get_cached, - invalidate_groups as _invalidate_groups, - invalidates as _invalidates, - set_cached as _set_cached, - cached_call as _cached_call, -) - -logger = logging.getLogger(__name__) - - -# ============================================ -# CONVERSATION-LEVEL CACHE (thin wrappers) -# ============================================ - -_NS = "_cache" # namespace for DEX cache - - -def get_cached(user_data: dict, key: str, ttl: int = DEFAULT_CACHE_TTL) -> Optional[Any]: - return _get_cached(user_data, key, ttl, namespace=_NS) - - -def set_cached(user_data: dict, key: str, value: Any) -> None: - _set_cached(user_data, key, value, namespace=_NS) - - -def evict_expired(user_data: dict) -> int: - return _evict_expired(user_data, namespace=_NS) - - -def clear_cache(user_data: dict, key: Optional[str] = None) -> None: - _clear_cache(user_data, key, namespace=_NS) - - -async def cached_call( - user_data: dict, - key: str, - fetch_func: Callable, - ttl: int = DEFAULT_CACHE_TTL, - *args, - **kwargs, -) -> Any: - return await _cached_call(user_data, key, fetch_func, ttl, *args, namespace=_NS, **kwargs) - - -# ============================================ -# CACHE INVALIDATION GROUPS -# ============================================ - -CACHE_GROUPS = { - "balances": [ - "gateway_balances", - "portfolio_data", - "wallet_balances", - "token_balances", - "gateway_data", - ], - "positions": [ - "clmm_positions", - "liquidity_positions", - "pool_positions", - "gateway_lp_positions", - "gateway_closed_positions", - ], - "swaps": ["swap_history", "recent_swaps"], - "tokens": ["token_cache"], - "all": None, -} - - -def invalidate_cache(user_data: dict, *groups: str) -> None: - """Invalidate cache keys by group name(s).""" - # Handle special direct-on-user_data keys for backward compat - for group in groups: - if group == "all": - user_data.pop("token_cache", None) - else: - keys = CACHE_GROUPS.get(group, [group]) - if keys: - for key in keys: - if key in user_data: - user_data.pop(key, None) - _invalidate_groups(user_data, CACHE_GROUPS, *groups, namespace=_NS) - - -def invalidates(*groups: str): - """Decorator that invalidates cache groups after handler execution.""" - return _invalidates(*groups, groups_map=CACHE_GROUPS, namespace=_NS) - - -# ============================================ -# SERVER CLIENT HELPERS -# ============================================ - -from config_manager import get_client - -# ============================================ -# EXPLORER URL GENERATION -# ============================================ - -SOLANA_EXPLORERS = { - "orb": "https://orb.helius.dev/tx/{tx_hash}?cluster={cluster}&tab=summary", - "solscan": "https://solscan.io/tx/{tx_hash}", - "solana_explorer": "https://explorer.solana.com/tx/{tx_hash}", -} - -ETHEREUM_EXPLORERS = { - "etherscan": "https://etherscan.io/tx/{tx_hash}", - "arbiscan": "https://arbiscan.io/tx/{tx_hash}", - "basescan": "https://basescan.org/tx/{tx_hash}", -} - - -def get_explorer_url(tx_hash: str, network: str) -> Optional[str]: - """Generate explorer URL for a transaction - - Args: - tx_hash: Transaction hash/signature - network: Network name (e.g., 'solana-mainnet-beta', 'ethereum-mainnet') - - Returns: - Explorer URL or None if network not supported - """ - if not tx_hash: - return None - - if network.startswith("solana"): - # Use Orb explorer for Solana (Helius) - cluster = "mainnet-beta" if "mainnet" in network else "devnet" - return SOLANA_EXPLORERS["orb"].format(tx_hash=tx_hash, cluster=cluster) - elif "ethereum" in network or "mainnet" in network: - if "arbitrum" in network: - return ETHEREUM_EXPLORERS["arbiscan"].format(tx_hash=tx_hash) - elif "base" in network: - return ETHEREUM_EXPLORERS["basescan"].format(tx_hash=tx_hash) - else: - return ETHEREUM_EXPLORERS["etherscan"].format(tx_hash=tx_hash) - - return None - - -def get_explorer_name(network: str) -> str: - """Get the explorer name for display - - Args: - network: Network name - - Returns: - Explorer name (e.g., 'Orb', 'Etherscan') - """ - if network.startswith("solana"): - return "Orb" - elif "arbitrum" in network: - return "Arbiscan" - elif "base" in network: - return "Basescan" - elif "ethereum" in network: - return "Etherscan" - return "Explorer" - - -# ============================================ -# SWAP FORMATTERS -# ============================================ - - -def format_swap_summary(swap: Dict[str, Any], include_explorer: bool = True) -> str: - """Format a swap record for display - - Args: - swap: Swap data dictionary - include_explorer: Whether to include explorer link - - Returns: - Formatted swap summary string (not escaped) - """ - pair = swap.get("trading_pair", "N/A") - side = swap.get("side", "N/A") - status = swap.get("status", "N/A") - network = swap.get("network", "") - tx_hash = swap.get("transaction_hash", "") - - # Format amounts - input_amount = swap.get("input_amount") - output_amount = swap.get("output_amount") - base_token = swap.get("base_token", "") - quote_token = swap.get("quote_token", "") - - # Build amount string - if input_amount is not None and output_amount is not None: - if side == "BUY": - # Buying base with quote - amount_str = f"{_format_amount(output_amount)} {base_token} for {_format_amount(input_amount)} {quote_token}" - else: - # Selling base for quote - amount_str = f"{_format_amount(input_amount)} {base_token} for {_format_amount(output_amount)} {quote_token}" - elif input_amount is not None: - amount_str = f"{_format_amount(input_amount)}" - else: - amount_str = "N/A" - - # Format price - price = swap.get("price") - price_str = f"@ {_format_price(price)}" if price else "" - - # Build the line - parts = [f"{side} {pair}", amount_str] - if price_str: - parts.append(price_str) - parts.append(f"[{status}]") - - return " ".join(parts) - - -def format_swap_detail(swap: Dict[str, Any]) -> str: - """Format detailed swap information - - Args: - swap: Swap data dictionary - - Returns: - Formatted multi-line swap details (not escaped) - """ - lines = [] - - # Header with status emoji - status = swap.get("status", "UNKNOWN") - status_emoji = get_status_emoji(status) - lines.append(f"{status_emoji} Swap Details") - lines.append("") - - # Trading info - pair = swap.get("trading_pair", "N/A") - side = swap.get("side", "N/A") - lines.append(f"Pair: {pair}") - lines.append(f"Side: {side}") - - # Amounts - input_amount = swap.get("input_amount") - output_amount = swap.get("output_amount") - base_token = swap.get("base_token", "") - quote_token = swap.get("quote_token", "") - - if input_amount is not None: - lines.append( - f"Input: {_format_amount(input_amount)} {quote_token if side == 'BUY' else base_token}" - ) - if output_amount is not None: - lines.append( - f"Output: {_format_amount(output_amount)} {base_token if side == 'BUY' else quote_token}" - ) - - # Price - price = swap.get("price") - if price: - lines.append(f"Price: {_format_price(price)}") - - # Slippage - slippage = swap.get("slippage_pct") - if slippage is not None: - lines.append(f"Slippage: {slippage}%") - - # Network info - lines.append("") - connector = swap.get("connector", "N/A") - network = swap.get("network", "N/A") - lines.append(f"Connector: {connector}") - lines.append(f"Network: {network}") - - # Transaction - tx_hash = swap.get("transaction_hash", "") - if tx_hash: - lines.append(f"Tx: {tx_hash[:16]}...") - - # Timestamp - timestamp = swap.get("timestamp", "") - if timestamp: - # Format timestamp for display - if "T" in timestamp: - date_part = timestamp.split("T")[0] - time_part = ( - timestamp.split("T")[1].split(".")[0] - if "." in timestamp.split("T")[1] - else timestamp.split("T")[1].split("+")[0] - ) - lines.append(f"Time: {date_part} {time_part}") - - # Status - lines.append(f"Status: {status}") - - return "\n".join(lines) - - -def get_status_emoji(status: str) -> str: - """Get emoji for swap status - - Args: - status: Status string (CONFIRMED, PENDING, FAILED, etc.) - - Returns: - Emoji character - """ - status_emojis = { - "CONFIRMED": "✅", - "PENDING": "⏳", - "FAILED": "❌", - "REJECTED": "🚫", - "UNKNOWN": "❓", - } - return status_emojis.get(status.upper(), "📊") - - -def _format_amount(amount: float) -> str: - """Format amount with appropriate precision""" - if amount is None: - return "N/A" - - if amount == 0: - return "0" - - # Use appropriate decimal places based on size - if abs(amount) >= 1000: - return f"{amount:,.2f}" - elif abs(amount) >= 1: - return f"{amount:.4f}" - elif abs(amount) >= 0.0001: - return f"{amount:.6f}" - else: - return f"{amount:.8f}" - - -def _format_price(price: float) -> str: - """Format price with appropriate precision""" - if price is None: - return "N/A" - - if price == 0: - return "0" - - if abs(price) >= 1: - return f"{price:.4f}" - elif abs(price) >= 0.0001: - return f"{price:.6f}" - else: - return f"{price:.10f}" - - -# ============================================ -# RELATIVE TIME FORMATTER -# ============================================ - - -def format_relative_time(timestamp: str) -> str: - """Format timestamp as relative time (e.g., '53s', '22m', '1h', '2d') - - Args: - timestamp: ISO format timestamp string - - Returns: - Relative time string - """ - from datetime import datetime, timezone - - if not timestamp: - return "" - - try: - # Parse ISO timestamp - if "T" in timestamp: - # Handle various ISO formats - ts_str = timestamp.replace("Z", "+00:00") - if "." in ts_str: - # Remove microseconds if present - parts = ts_str.split(".") - if "+" in parts[1]: - ts_str = parts[0] + "+" + parts[1].split("+")[1] - elif "-" in parts[1]: - ts_str = parts[0] + "-" + parts[1].split("-", 1)[1] - else: - ts_str = parts[0] - - # Parse with timezone - try: - dt = datetime.fromisoformat(ts_str) - except ValueError: - # Fallback: try without timezone - dt = datetime.fromisoformat(timestamp.split("+")[0].split(".")[0]) - dt = dt.replace(tzinfo=timezone.utc) - else: - return "" - - # Calculate difference - now = datetime.now(timezone.utc) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - - diff = now - dt - seconds = int(diff.total_seconds()) - - if seconds < 0: - return "now" - elif seconds < 60: - return f"{seconds}s" - elif seconds < 3600: - return f"{seconds // 60}m" - elif seconds < 86400: - return f"{seconds // 3600}h" - else: - return f"{seconds // 86400}d" - - except Exception as e: - logger.debug(f"Error formatting relative time: {e}") - return "" - - -# ============================================ -# STATE HELPERS -# ============================================ - - -def clear_dex_state(context) -> None: - """Clear all DEX-related state from user context - - Args: - context: Telegram context object - """ - context.user_data.pop("dex_state", None) - context.user_data.pop("dex_previous_state", None) - context.user_data.pop("quote_swap_params", None) - context.user_data.pop("execute_swap_params", None) - - -# ============================================ -# HISTORY FILTER & PAGINATION HELPERS -# ============================================ - -from dataclasses import dataclass -from typing import Literal - -HistoryType = Literal["swap", "position"] - -# Available filter options per history type -HISTORY_FILTERS = { - "swap": { - "trading_pair": ["All", "SOL-USDC", "SOL-ORE", "ORE-USDC", "ETH-USDC"], - "connector": ["All", "jupiter", "uniswap"], - "status": ["All", "CONFIRMED", "PENDING", "FAILED"], - }, - "position": { - "trading_pair": ["All", "SOL-USDC", "ORE-SOL", "METv-SOL"], - "connector": ["All", "meteora", "orca", "raydium"], - "status": ["All", "OPEN", "CLOSED"], - }, -} - -DEFAULT_PAGE_SIZE = 10 - - -@dataclass -class HistoryFilters: - """Stores filter and pagination state for history views""" - - history_type: HistoryType = "swap" - trading_pair: Optional[str] = None # None = All - connector: Optional[str] = None # None = All - status: Optional[str] = None # None = All - network: Optional[str] = None # None = All - offset: int = 0 - limit: int = DEFAULT_PAGE_SIZE - total_count: int = 0 - - def to_dict(self) -> Dict[str, Any]: - return { - "history_type": self.history_type, - "trading_pair": self.trading_pair, - "connector": self.connector, - "status": self.status, - "network": self.network, - "offset": self.offset, - "limit": self.limit, - "total_count": self.total_count, - } - - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "HistoryFilters": - return cls( - history_type=data.get("history_type", "swap"), - trading_pair=data.get("trading_pair"), - connector=data.get("connector"), - status=data.get("status"), - network=data.get("network"), - offset=data.get("offset", 0), - limit=data.get("limit", DEFAULT_PAGE_SIZE), - total_count=data.get("total_count", 0), - ) - - def reset_pagination(self) -> None: - """Reset pagination when filters change""" - self.offset = 0 - - @property - def current_page(self) -> int: - return (self.offset // self.limit) + 1 - - @property - def total_pages(self) -> int: - if self.total_count == 0: - return 1 - return (self.total_count + self.limit - 1) // self.limit - - @property - def has_next(self) -> bool: - return self.offset + self.limit < self.total_count - - @property - def has_prev(self) -> bool: - return self.offset > 0 - - -def get_history_filters(user_data: dict, history_type: HistoryType) -> HistoryFilters: - """Get current history filters from user data""" - key = f"history_filters_{history_type}" - data = user_data.get(key) - if data: - return HistoryFilters.from_dict(data) - return HistoryFilters(history_type=history_type) - - -def set_history_filters(user_data: dict, filters: HistoryFilters) -> None: - """Save history filters to user data""" - key = f"history_filters_{filters.history_type}" - user_data[key] = filters.to_dict() - - -def build_filter_buttons( - filters: HistoryFilters, callback_prefix: str -) -> List[List["InlineKeyboardButton"]]: - """Build filter button rows for history views - - Args: - filters: Current filter state - callback_prefix: Prefix for callback data (e.g., "dex:swap_hist" or "dex:lp_hist") - - Returns: - List of button rows - """ - from telegram import InlineKeyboardButton - - rows = [] - - # Trading pair filter - pair_label = filters.trading_pair or "All Pairs" - rows.append( - [ - InlineKeyboardButton( - f"💱 {pair_label}", callback_data=f"{callback_prefix}_filter_pair" - ), - ] - ) - - # Connector & Status filters (same row) - connector_label = filters.connector or "All DEX" - status_label = filters.status or "All Status" - rows.append( - [ - InlineKeyboardButton( - f"🔌 {connector_label}", - callback_data=f"{callback_prefix}_filter_connector", - ), - InlineKeyboardButton( - f"📊 {status_label}", callback_data=f"{callback_prefix}_filter_status" - ), - ] - ) - - return rows - - -def build_pagination_buttons( - filters: HistoryFilters, callback_prefix: str -) -> List["InlineKeyboardButton"]: - """Build pagination buttons for history views - - Args: - filters: Current filter state with pagination info - callback_prefix: Prefix for callback data - - Returns: - List of buttons for a single row - """ - from telegram import InlineKeyboardButton - - buttons = [] - - # Previous button - if filters.has_prev: - buttons.append( - InlineKeyboardButton("« Prev", callback_data=f"{callback_prefix}_page_prev") - ) - else: - buttons.append(InlineKeyboardButton(" ", callback_data="dex:noop")) - - # Page indicator - page_text = f"{filters.current_page}/{filters.total_pages}" - buttons.append(InlineKeyboardButton(page_text, callback_data="dex:noop")) - - # Next button - if filters.has_next: - buttons.append( - InlineKeyboardButton("Next »", callback_data=f"{callback_prefix}_page_next") - ) - else: - buttons.append(InlineKeyboardButton(" ", callback_data="dex:noop")) - - return buttons - - -def build_filter_selection_keyboard( - options: List[str], - current_value: Optional[str], - callback_prefix: str, - back_callback: str, -) -> "InlineKeyboardMarkup": - """Build a keyboard for selecting a filter value - - Args: - options: List of available options - current_value: Currently selected value (None = All) - callback_prefix: Prefix for callback data - back_callback: Callback for back button - - Returns: - InlineKeyboardMarkup with option buttons - """ - from telegram import InlineKeyboardButton, InlineKeyboardMarkup - - buttons = [] - row = [] - - for opt in options: - # Check if this option is currently selected - is_selected = (opt == "All" and current_value is None) or (opt == current_value) - label = f"✓ {opt}" if is_selected else opt - - # Use None for "All" option - value = "" if opt == "All" else opt - row.append( - InlineKeyboardButton(label, callback_data=f"{callback_prefix}_{value}") - ) - - if len(row) == 2: - buttons.append(row) - row = [] - - if row: - buttons.append(row) - - buttons.append([InlineKeyboardButton("« Back", callback_data=back_callback)]) - - return InlineKeyboardMarkup(buttons) diff --git a/market.py b/market.py deleted file mode 100644 index d6c8c732..00000000 --- a/market.py +++ /dev/null @@ -1,611 +0,0 @@ -from __future__ import annotations - -import logging -import re -import time - -from fastapi import APIRouter, Depends, HTTPException, Query - -from config_manager import get_config_manager - -logger = logging.getLogger(__name__) - -# Simple TTL cache for candle data -_candle_cache: dict[tuple, tuple[float, list]] = {} # key -> (timestamp, data) -_CANDLE_CACHE_TTL = 30.0 # seconds -_CANDLE_CACHE_MAX = 50 # hard cap on entries (keys rotate every minute per chart) - -# Persistent dict handed to handlers.dex.pool_data.fetch_ohlcv so its own 300s -# GeckoTerminal cache applies on top of the 30s route cache above — keeps us well -# under GeckoTerminal's free-tier rate limit when multiple pool charts are open. -_gecko_ohlcv_user_data: dict = {} - - -def _candle_cache_put(key: tuple, value: list, now: float) -> None: - """Insert into the candle cache, evicting expired entries and capping size.""" - expired = [ - k for k, (ts, _) in _candle_cache.items() if now - ts >= _CANDLE_CACHE_TTL - ] - for k in expired: - _candle_cache.pop(k, None) - _candle_cache[key] = (now, value) - while len(_candle_cache) > _CANDLE_CACHE_MAX: - # dicts preserve insertion order: drop the oldest entry first - _candle_cache.pop(next(iter(_candle_cache))) - - -from condor.web.auth import get_current_user -from condor.web.models import ( - CandleData, - MarketPriceResponse, - OrderBookLevel, - OrderBookResponse, - TradingRuleItem, - TradingRulesResponse, - WebUser, -) - -router = APIRouter(tags=["market"]) - - -_MINT_RE = re.compile(r"^[1-9A-HJ-NP-Za-km-z]{32,44}$") - -# DEX pool addresses across GeckoTerminal networks: base58 (Solana) or 0x-hex -# (EVM). Used to sanitize the pool_address query param before it reaches a URL. -_POOL_ADDR_RE = re.compile(r"^[A-Za-z0-9]{16,90}$") - - -async def _fetch_pool_candles_raw( - pool_address: str, - network: str, - interval: str, - limit: int = 100, - before_timestamp: int | None = None, -) -> list[CandleData]: - """OHLCV rows for one pool from GeckoTerminal (reuses handlers.dex fetch+cache). - - ``currency="token"`` prices the base token in the quote token (e.g. SOL), matching - the executor's own entry/range price scale drawn on the same chart. ``limit`` and - ``before_timestamp`` carry the chart's requested window so an archived executor - charts against the candles it actually traded in, not the latest ones. Returns [] - on any miss/error (never raises) so DEX pairs don't fall to the CEX 502 path. - """ - from handlers.dex.pool_data import fetch_ohlcv - - try: - ohlcv_list, err = await fetch_ohlcv( - pool_address, - network, - timeframe=interval, - currency="token", - user_data=_gecko_ohlcv_user_data, - limit=limit, - before_timestamp=before_timestamp, - ) - except Exception as e: - logger.warning( - "GeckoTerminal OHLCV failed pool=%s net=%s interval=%s: %s", - pool_address, - network, - interval, - e, - ) - return [] - if err or not ohlcv_list: - return [] - - candles: list[CandleData] = [] - for c in ohlcv_list: - # Rows are [timestamp, open, high, low, close, volume(_usd), (datetime)]. - if not isinstance(c, (list, tuple)) or len(c) < 6: - continue - try: - candles.append( - CandleData( - timestamp=float(c[0]), - open=float(c[1]), - high=float(c[2]), - low=float(c[3]), - close=float(c[4]), - volume=float(c[5]), - ) - ) - except (TypeError, ValueError): - continue - return candles - - -# Base-token mint → its top GeckoTerminal pool (24h-volume-sorted). Pools are stable, -# so cache for an hour. Lets an executor chart fall back to the token's live main pool -# when its own pool_address is stale/absent (e.g. a closed slot, or a multi-executor -# group where the chart picked a dead pool). -_token_pool_cache: dict[tuple[str, str], tuple[float, str]] = {} -_TOKEN_POOL_TTL = 3600.0 - - -async def _resolve_token_top_pool(mint: str, gnet: str, quote: str = "SOL") -> str: - key = (gnet, mint) - now = time.time() - cached = _token_pool_cache.get(key) - if cached and (now - cached[0]) < _TOKEN_POOL_TTL: - return cached[1] - - addr = "" - try: - import aiohttp - - url = f"https://api.geckoterminal.com/api/v2/networks/{gnet}/tokens/{mint}/pools?page=1" - async with aiohttp.ClientSession() as s: - async with s.get( - url, headers={"Accept": "application/json;version=20230302"} - ) as r: - r.raise_for_status() - data = await r.json() - pools = data.get("data") or [] - # Prefer a pool quoted in the executor's quote token (e.g. SOL) so the price - # scale matches; else the highest-volume pool (list is volume-sorted). - chosen = None - for p in pools: - parts = str((p.get("attributes") or {}).get("name") or "").upper().replace(" ", "").split("/") - if quote and quote.upper() in parts: - chosen = p - break - chosen = chosen or (pools[0] if pools else None) - if chosen: - attrs = chosen.get("attributes") or {} - addr = str(attrs.get("address") or str(chosen.get("id") or "").split("_")[-1] or "") - # Only cache on a successful API response — addr "" here means the token - # genuinely has no pool, which is worth caching. A transient error (below) - # must not poison the cache for an hour, so it returns without caching. - _token_pool_cache[key] = (now, addr) - return addr - except Exception as e: - logger.info("top-pool resolve failed mint=%s net=%s: %s", mint, gnet, e) - return "" - - -async def _get_pool_candles( - connector: str, - pool_address: str | None, - trading_pair: str, - interval: str, - cache_key: tuple, - now: float, - limit: int = 100, - before_timestamp: int | None = None, -) -> list[CandleData]: - """Candles for a DEX/LP pair from GeckoTerminal. - - Tries the executor's own ``pool_address`` first (exact pool); if that yields - nothing — a stale/closed slot, no pool_address, or a group whose first executor - sits on a dead pool — falls back to the base token's top live pool resolved from - the mint in ``trading_pair``. So a live token always charts even when the passed - pool is wrong. ``connector`` is the network id (e.g. solana-mainnet-beta). - ``limit``/``before_timestamp`` carry the chart's requested window (see - :func:`_fetch_pool_candles_raw`). - """ - from handlers.dex.pool_data import get_gecko_network - - gnet = get_gecko_network(connector) - candles: list[CandleData] = [] - if pool_address: - candles = await _fetch_pool_candles_raw( - pool_address, connector, interval, limit, before_timestamp - ) - - if not candles: - dash = trading_pair.rfind("-") - base = trading_pair[:dash] if dash > 0 else trading_pair - quote = trading_pair[dash + 1 :] if dash > 0 else "SOL" - if _MINT_RE.match(base): - top = await _resolve_token_top_pool(base, gnet, quote) - if top and top != pool_address: - candles = await _fetch_pool_candles_raw( - top, connector, interval, limit, before_timestamp - ) - - _candle_cache_put(cache_key, candles, now) - return candles - - -@router.get("/servers/{name}/market/connectors") -async def get_connectors(name: str, user: WebUser = Depends(get_current_user)): - cm = get_config_manager() - if not cm.has_server_access(user.id, name): - raise HTTPException(status_code=403, detail="No access") - - from condor.server_data_service import ServerDataType, get_server_data_service - - try: - result = await get_server_data_service().get_or_fetch( - name, ServerDataType.CANDLE_CONNECTORS - ) - except Exception as e: - raise HTTPException(status_code=502, detail=str(e)) - return result - - -@router.get("/servers/{name}/market/connected-exchanges") -async def get_connected_exchanges(name: str, user: WebUser = Depends(get_current_user)): - """Get connectors that have credentials configured (accounts connected).""" - cm = get_config_manager() - if not cm.has_server_access(user.id, name): - raise HTTPException(status_code=403, detail="No access") - - from condor.server_data_service import ServerDataType, get_server_data_service - - try: - result = await get_server_data_service().get_or_fetch( - name, ServerDataType.CONNECTORS - ) - except Exception as e: - raise HTTPException(status_code=502, detail=str(e)) - return result or [] - - -@router.get("/servers/{name}/market/prices", response_model=MarketPriceResponse) -async def get_price( - name: str, - connector: str = Query(...), - trading_pair: str = Query(...), - user: WebUser = Depends(get_current_user), -): - cm = get_config_manager() - if not cm.has_server_access(user.id, name): - raise HTTPException(status_code=403, detail="No access") - - from condor.server_data_service import ServerDataType, get_server_data_service - - try: - result = await get_server_data_service().get_or_fetch( - name, - ServerDataType.PRICES, - connector_name=connector, - trading_pair=trading_pair, - ) - except Exception as e: - raise HTTPException(status_code=502, detail=str(e)) - - if result is None: - raise HTTPException(status_code=502, detail="Failed to fetch price") - - if isinstance(result, (int, float)): - return MarketPriceResponse( - connector=connector, trading_pair=trading_pair, mid_price=float(result) - ) - elif isinstance(result, dict): - return MarketPriceResponse( - connector=connector, - trading_pair=trading_pair, - mid_price=float(result.get("mid_price", result.get("price", 0))), - best_bid=float(result.get("best_bid", 0)), - best_ask=float(result.get("best_ask", 0)), - ) - raise HTTPException(status_code=502, detail="Unexpected response format") - - -@router.post("/servers/{name}/rate-oracle/rates") -async def get_rate_oracle_rates( - name: str, - body: dict, - user: WebUser = Depends(get_current_user), -): - cm = get_config_manager() - if not cm.has_server_access(user.id, name): - raise HTTPException(status_code=403, detail="No access") - - trading_pairs = body.get("trading_pairs", []) - if not trading_pairs: - return {"rates": {}} - - client = await cm.get_client(name) - try: - result = await client.rate_oracle.get_rates(trading_pairs=trading_pairs) - except Exception as e: - raise HTTPException(status_code=502, detail=str(e)) - - return result - - -@router.get("/servers/{name}/market/trading-rules", response_model=TradingRulesResponse) -async def get_trading_rules( - name: str, - connector: str = Query(...), - user: WebUser = Depends(get_current_user), -): - cm = get_config_manager() - if not cm.has_server_access(user.id, name): - raise HTTPException(status_code=403, detail="No access") - - from condor.server_data_service import ServerDataType, get_server_data_service - - try: - result = await get_server_data_service().get_or_fetch( - name, ServerDataType.TRADING_RULES, connector_name=connector - ) - except Exception as e: - raise HTTPException(status_code=502, detail=str(e)) - - if not isinstance(result, dict): - return TradingRulesResponse(connector=connector, rules=[]) - - rules = [] - for pair, rule_data in result.items(): - if isinstance(rule_data, dict): - rules.append( - TradingRuleItem( - trading_pair=pair, - min_order_size=float(rule_data.get("min_order_size", 0)), - min_notional_size=float(rule_data.get("min_notional_size", 0)), - min_price_increment=float(rule_data.get("min_price_increment", 0)), - min_base_amount_increment=float( - rule_data.get("min_base_amount_increment", 0) - ), - ) - ) - return TradingRulesResponse(connector=connector, rules=rules) - - -@router.get("/servers/{name}/market/order-book", response_model=OrderBookResponse) -async def get_order_book( - name: str, - connector: str = Query(...), - trading_pair: str = Query(...), - depth: int = Query(default=20, ge=1, le=100), - user: WebUser = Depends(get_current_user), -): - cm = get_config_manager() - if not cm.has_server_access(user.id, name): - raise HTTPException(status_code=403, detail="No access") - - client = await cm.get_client(name) - try: - result = await client.market_data.get_order_book( - connector_name=connector, trading_pair=trading_pair - ) - except Exception as e: - raise HTTPException(status_code=502, detail=str(e)) - - bids = [] - asks = [] - if isinstance(result, dict): - for entry in (result.get("bids") or [])[:depth]: - if isinstance(entry, (list, tuple)) and len(entry) >= 2: - bids.append( - OrderBookLevel(price=float(entry[0]), amount=float(entry[1])) - ) - elif isinstance(entry, dict): - bids.append( - OrderBookLevel( - price=float(entry.get("price", 0)), - amount=float(entry.get("amount", entry.get("quantity", 0))), - ) - ) - for entry in (result.get("asks") or [])[:depth]: - if isinstance(entry, (list, tuple)) and len(entry) >= 2: - asks.append( - OrderBookLevel(price=float(entry[0]), amount=float(entry[1])) - ) - elif isinstance(entry, dict): - asks.append( - OrderBookLevel( - price=float(entry.get("price", 0)), - amount=float(entry.get("amount", entry.get("quantity", 0))), - ) - ) - - return OrderBookResponse( - connector=connector, trading_pair=trading_pair, bids=bids, asks=asks - ) - - -@router.get("/servers/{name}/market/candles", response_model=list[CandleData]) -async def get_candles( - name: str, - connector: str = Query(...), - trading_pair: str = Query(...), - interval: str = Query(default="1m"), - limit: int = Query(default=1000, ge=1, le=5000), - start_time: float | None = Query(default=None, description="Unix epoch seconds"), - end_time: float | None = Query(default=None, description="Unix epoch seconds"), - pool_address: str | None = Query( - default=None, - description="DEX pool address. When set, candles are fetched from " - "GeckoTerminal (by pool) instead of the CEX candle feed — used for LP/DEX " - "executors whose connector (e.g. solana-mainnet-beta) has no CandlesFactory feed.", - ), - user: WebUser = Depends(get_current_user), -): - cm = get_config_manager() - if not cm.has_server_access(user.id, name): - raise HTTPException(status_code=403, detail="No access") - - # pool_address is interpolated into GeckoTerminal URLs — restrict to plain - # address characters (base58 for Solana, 0x-hex for EVM networks). - if pool_address and not _POOL_ADDR_RE.match(pool_address): - raise HTTPException(status_code=400, detail="Invalid pool_address") - - # Bucket start_time to 60s intervals so near-identical requests share cache - bucketed_start = int(start_time // 60) * 60 if start_time is not None else None - bucketed_end = int(end_time // 60) * 60 if end_time is not None else None - cache_key = ( - name, - connector, - trading_pair, - interval, - limit, - bucketed_start, - bucketed_end, - pool_address, - ) - now = time.monotonic() - cached = _candle_cache.get(cache_key) - if cached and (now - cached[0]) < _CANDLE_CACHE_TTL: - return cached[1] - - # DEX/LP pools have no CEX candle feed — route to GeckoTerminal. Trigger on a - # DEX network connector (e.g. "solana-mainnet-beta") OR an explicit pool_address, - # so these pairs never fall through to the CEX path (which 502s). _get_pool_candles - # uses the pool_address when it has data, else resolves the token's top pool. - from handlers.dex.pool_data import NETWORK_TO_GECKO - - if pool_address or connector in NETWORK_TO_GECKO: - # Pass the chart's window through so archived executors chart against the - # candles they actually traded in. before_timestamp = end of window (candles - # walk back from there); None = latest. GeckoTerminal caps limit at 1000. - return await _get_pool_candles( - connector, - pool_address, - trading_pair, - interval, - cache_key, - now, - limit=limit, - before_timestamp=bucketed_end, - ) - - client = await cm.get_client(name) - result = None - try: - # Prefer historical candles with time range when start_time is given - if start_time is not None: - st = int(start_time) - et = int(end_time) if end_time else int(time.time()) - logger.info( - "Fetching historical candles: connector=%s pair=%s interval=%s start=%s end=%s", - connector, - trading_pair, - interval, - st, - et, - ) - result = await client.market_data.get_historical_candles( - connector, - trading_pair, - interval, - start_time=st, - end_time=et, - ) - logger.info( - "Historical candles result: type=%s len=%s", - type(result).__name__, - len(result) if isinstance(result, (list, dict)) else "?", - ) - except Exception as e: - logger.warning( - "get_historical_candles failed: %s — falling back to get_candles", e - ) - result = None - - # Fallback: if historical returned nothing usable, use regular candles - candles_raw = ( - result - if isinstance(result, list) - else result.get("data", []) if isinstance(result, dict) else [] - ) - if not candles_raw: - try: - logger.info( - "Falling back to get_candles: connector=%s pair=%s interval=%s limit=%s", - connector, - trading_pair, - interval, - limit, - ) - result = await client.market_data.get_candles( - connector, trading_pair, interval, limit - ) - except Exception as e: - raise HTTPException(status_code=502, detail=str(e)) - - candles_raw = ( - result - if isinstance(result, list) - else result.get("data", []) if isinstance(result, dict) else [] - ) - - candles = [] - for c in candles_raw: - if isinstance(c, dict): - candles.append( - CandleData( - timestamp=float(c.get("timestamp", 0)), - open=float(c.get("open", 0)), - high=float(c.get("high", 0)), - low=float(c.get("low", 0)), - close=float(c.get("close", 0)), - volume=float(c.get("volume", 0)), - ) - ) - elif isinstance(c, (list, tuple)) and len(c) >= 6: - candles.append( - CandleData( - timestamp=float(c[0]), - open=float(c[1]), - high=float(c[2]), - low=float(c[3]), - close=float(c[4]), - volume=float(c[5]), - ) - ) - _candle_cache_put(cache_key, candles, now) - return candles - - -# Token symbol resolution — LP/DEX executors store `trading_pair` as `-SOL` -# (Gateway can't resolve memecoins by symbol), so the dashboard shows the raw mint. -# Resolve mint → ticker via GeckoTerminal (same source as candles). Symbols are -# stable, so cache for a day. Empty string is cached too (so an unknown/illiquid -# mint doesn't re-hit GeckoTerminal every render); the UI falls back to the mint. -_token_symbol_cache: dict[tuple[str, str], tuple[float, str]] = {} -_TOKEN_SYMBOL_TTL = 24 * 3600.0 - - -@router.get("/market/token-symbol") -async def get_token_symbol( - mint: str = Query(..., description="Base token mint address"), - network: str = Query( - default="solana", description="Network id or connector (e.g. solana-mainnet-beta)" - ), - user: WebUser = Depends(get_current_user), -): - # Server-independent: pure GeckoTerminal lookup, no server scoping needed - # (auth still required). Lets the executor tables resolve symbols without - # threading a server name into every row. - from handlers.dex.pool_data import get_gecko_network - - # The mint is interpolated into the GeckoTerminal URL path — reject anything - # that isn't a base58 pubkey (mirrors the frontend's looksLikeMint gate). - if not _MINT_RE.match(mint): - raise HTTPException(status_code=400, detail="Invalid mint address") - - gnet = get_gecko_network(network) - key = (gnet, mint) - now = time.time() - cached = _token_symbol_cache.get(key) - if cached and (now - cached[0]) < _TOKEN_SYMBOL_TTL: - return {"mint": mint, "symbol": cached[1]} - - symbol = "" - try: - import aiohttp - - url = f"https://api.geckoterminal.com/api/v2/networks/{gnet}/tokens/{mint}" - async with aiohttp.ClientSession() as s: - async with s.get( - url, headers={"Accept": "application/json;version=20230302"} - ) as r: - r.raise_for_status() - data = await r.json() - symbol = str( - (((data or {}).get("data") or {}).get("attributes") or {}).get("symbol") or "" - ) - except Exception as e: - # Don't cache a transient failure — a single blip must not blank this pair's - # ticker for 24h. Only successful responses (below) are cached, empty included - # (a genuinely unknown mint is worth remembering). - logger.info("token-symbol resolve failed for mint=%s network=%s: %s", mint, gnet, e) - return {"mint": mint, "symbol": ""} - - _token_symbol_cache[key] = (now, symbol) - return {"mint": mint, "symbol": symbol} diff --git a/pool_data.py b/pool_data.py deleted file mode 100644 index 815cf5dd..00000000 --- a/pool_data.py +++ /dev/null @@ -1,465 +0,0 @@ -""" -Pool Data Utilities - -Provides unified data fetching for DEX pools: -- OHLCV data via GeckoTerminal (works for any pool on any DEX) -- Liquidity/bin data via Gateway CLMM (for supported DEXes) -- Pool info normalization across different sources -""" - -import logging -from typing import Any, Dict, List, Optional, Tuple - -from geckoterminal_py import GeckoTerminalAsyncClient - -from config_manager import get_client - -from ._shared import evict_expired, get_cached, set_cached - -logger = logging.getLogger(__name__) - -# Supported DEXes for liquidity data (via gateway CLMM) -LIQUIDITY_SUPPORTED_DEXES = { - "meteora": "solana", - "raydium": "solana", - "orca": "solana", - "uniswap": "ethereum", - "pancakeswap": "bsc", -} - -# GeckoTerminal network mapping -NETWORK_TO_GECKO = { - "solana": "solana", - "solana-mainnet-beta": "solana", - "ethereum": "eth", - "ethereum-mainnet": "eth", - "arbitrum": "arbitrum", - "arbitrum-one": "arbitrum", - "base": "base", - "base-mainnet": "base", - "bsc": "bsc", - "binance-smart-chain": "bsc", - "polygon": "polygon_pos", - "polygon-mainnet": "polygon_pos", - "avalanche": "avalanche", - "optimism": "optimism", -} - -# DEX ID to GeckoTerminal DEX mapping -DEX_TO_GECKO = { - "meteora": "meteora", - "raydium": "raydium", - "orca": "orca", - "uniswap": "uniswap", - "uniswap_v3": "uniswap_v3", - "pancakeswap": "pancakeswap", - "pancakeswap_v3": "pancakeswap_v3", - "sushiswap": "sushiswap", -} - -# Cache TTLs -OHLCV_CACHE_TTL = 300 # 5 minutes -BINS_CACHE_TTL = 60 # 1 minute - - -def get_gecko_network(network: str) -> str: - """Convert internal network name to GeckoTerminal network ID""" - return NETWORK_TO_GECKO.get(network, network) - - -def can_fetch_liquidity(dex_id: str, network: str = None) -> bool: - """Check if liquidity/bin data can be fetched for this DEX - - Args: - dex_id: DEX identifier (e.g., "meteora", "raydium") - network: Optional network to verify (must be Solana for now) - - Returns: - True if liquidity data is available via gateway CLMM - """ - dex_lower = dex_id.lower() if dex_id else "" - - if dex_lower not in LIQUIDITY_SUPPORTED_DEXES: - return False - - if network: - expected_network = LIQUIDITY_SUPPORTED_DEXES.get(dex_lower) - gecko_network = get_gecko_network(network) - if gecko_network != expected_network: - return False - - return True - - -def get_connector_for_dex(dex_id: str) -> Optional[str]: - """Get the gateway connector name for a DEX ID - - Args: - dex_id: DEX identifier from GeckoTerminal - - Returns: - Connector name for gateway CLMM or None - """ - dex_lower = dex_id.lower() if dex_id else "" - - # Direct mapping - if dex_lower in LIQUIDITY_SUPPORTED_DEXES: - return dex_lower - - # Handle variations - if "meteora" in dex_lower: - return "meteora" - if "raydium" in dex_lower: - return "raydium" - if "orca" in dex_lower: - return "orca" - - return None - - -async def fetch_ohlcv( - pool_address: str, - network: str, - timeframe: str = "1h", - currency: str = "usd", - user_data: dict = None, - limit: int = 100, - before_timestamp: Optional[int] = None, -) -> Tuple[Optional[List], Optional[str]]: - """Fetch OHLCV data for any pool via GeckoTerminal - - Args: - pool_address: Pool contract address - network: Network identifier (will be converted to GeckoTerminal format) - timeframe: OHLCV timeframe ("1m", "5m", "15m", "1h", "4h", "1d") - currency: Price currency - "usd" or "token" (quote token) - user_data: Optional user_data dict for caching - limit: Number of candles to fetch (GeckoTerminal caps at 1000) - before_timestamp: Fetch candles ending at this unix-seconds timestamp - (walks history back from here); None = latest candles. Needed so an - archived executor charts against the price window it actually traded - in, not the latest candles. - - Returns: - Tuple of (ohlcv_list, error_message) - ohlcv_list: List of [timestamp, open, high, low, close, volume] or None - error_message: Error string if failed, None on success - """ - try: - gecko_network = get_gecko_network(network) - # GeckoTerminal's OHLCV endpoint caps limit at 1000. - limit = max(1, min(int(limit), 1000)) - - # Check cache. before_timestamp/limit are part of the key so a historical - # window and the live window for the same pool don't collide. - if user_data is not None: - cache_key = ( - f"ohlcv_{gecko_network}_{pool_address}_{timeframe}_{currency}" - f"_{limit}_{before_timestamp or 0}" - ) - cached = get_cached(user_data, cache_key, ttl=OHLCV_CACHE_TTL) - if cached is not None: - return cached, None - # Sweep stale entries on every miss (same contract as cached_call). - # Keys now include limit/before_timestamp, which rotates by the minute - # for live charts — without eviction a long-lived process (the web - # route's persistent dict) grows without bound. - evict_expired(user_data) - - client = GeckoTerminalAsyncClient() - # Pass all parameters explicitly: - # - currency="token" means price in quote token (not USD) - # - token="base" means OHLCV for the base token - result = await client.get_ohlcv( - gecko_network, - pool_address, - timeframe, - before_timestamp=before_timestamp, - currency=currency, - token="base", - limit=limit, - ) - - # Parse response - handle different formats - ohlcv_list = None - - try: - import pandas as pd - - if isinstance(result, pd.DataFrame): - if not result.empty: - # Convert DataFrame to list format - ohlcv_list = result.values.tolist() - except ImportError: - pass - - if ohlcv_list is None: - if isinstance(result, list): - ohlcv_list = result - elif isinstance(result, dict): - # Try nested structure - data = result.get("data", result) - if isinstance(data, dict): - attrs = data.get("attributes", data) - ohlcv_list = attrs.get("ohlcv_list", []) - elif isinstance(data, list): - ohlcv_list = data - - if not ohlcv_list: - return None, "No OHLCV data available" - - # Debug logging: show price range from OHLCV data - if ohlcv_list: - try: - closes = [float(c[4]) for c in ohlcv_list if len(c) > 4 and c[4]] - if closes: - logger.info( - f"OHLCV {pool_address[:8]}... {timeframe} currency={currency}: " - f"{len(ohlcv_list)} candles, price range [{min(closes):.6f} - {max(closes):.6f}]" - ) - except Exception as e: - logger.debug(f"Could not log OHLCV price range: {e}") - - # Cache result - if user_data is not None: - set_cached(user_data, cache_key, ohlcv_list) - - return ohlcv_list, None - - except Exception as e: - logger.error(f"Error fetching OHLCV: {e}", exc_info=True) - return None, f"Failed to fetch OHLCV: {str(e)}" - - -async def fetch_liquidity_bins( - pool_address: str, - connector: str = "meteora", - network: str = "solana-mainnet-beta", - user_data: dict = None, - chat_id: int = None, - context=None, -) -> Tuple[Optional[List], Optional[Dict], Optional[str]]: - """Fetch liquidity bin data for CLMM pools via gateway - - Args: - pool_address: Pool contract address - connector: DEX connector (meteora, raydium, orca) - network: Network identifier - user_data: Optional user_data dict for caching - chat_id: Chat ID for per-chat server selection - - Returns: - Tuple of (bins_list, pool_info, error_message) - bins_list: List of bin dicts with price, base_token_amount, quote_token_amount - pool_info: Full pool info dict - error_message: Error string if failed, None on success - """ - try: - if not can_fetch_liquidity(connector): - return None, None, f"Liquidity data not available for {connector}" - - # Check cache - cache_key = f"pool_bins_{connector}_{pool_address}" - if user_data is not None: - cached = get_cached(user_data, cache_key, ttl=BINS_CACHE_TTL) - if cached is not None: - return cached.get("bins"), cached, None - - client = await get_client(chat_id, context=context) - if not client: - return None, None, "Gateway client not available" - - pool_info = None - - # First try get_pool_info (works for pools known to gateway) - try: - pool_info = await client.gateway_clmm.get_pool_info( - connector=connector, network=network, pool_address=pool_address - ) - except Exception as e: - # If get_pool_info fails (e.g., pool not in gateway config or not a DLMM pool), - # try finding the pool via get_pools search - error_str = str(e) - if "validation error" in error_str.lower() or "Field required" in error_str: - logger.info( - f"Pool {pool_address[:12]}... not found via get_pool_info, trying get_pools search" - ) - try: - # Search for pool by address using get_pools - search_result = await client.gateway_clmm.get_pools( - connector=connector, search_term=pool_address, limit=1 - ) - pools = search_result.get("pools", []) - if pools: - # Found the pool, but get_pools doesn't include bins - # Return pool info without bins - caller can handle this - pool_info = pools[0] - pool_info["address"] = pool_address - logger.info( - f"Found pool via get_pools: {pool_info.get('trading_pair', 'Unknown')}" - ) - else: - # Pool not found in DLMM pools - might be an AMM pool or non-existent - logger.info( - f"Pool {pool_address[:12]}... not found in {connector} DLMM pools" - ) - return ( - None, - None, - f"Pool not found in {connector} DLMM pools. This may be an AMM pool or not a {connector} pool.", - ) - except Exception as search_e: - logger.warning(f"get_pools search also failed: {search_e}") - return ( - None, - None, - f"Could not fetch pool info. Pool may not be a {connector} DLMM pool.", - ) - - if pool_info is None: - # Re-raise with a cleaner message for non-validation errors - return None, None, f"Failed to fetch pool: {str(e)[:100]}" - - if not pool_info: - return None, None, "Pool not found" - - bins = pool_info.get("bins", []) - - # Cache result - if user_data is not None: - set_cached(user_data, cache_key, pool_info) - - return bins, pool_info, None - - except Exception as e: - logger.error(f"Error fetching liquidity bins: {e}", exc_info=True) - return None, None, f"Failed to fetch liquidity: {str(e)}" - - -def normalize_pool_data(pool: dict, source: str = "gecko") -> Dict[str, Any]: - """Normalize pool data from different sources to a common format - - Args: - pool: Raw pool data dict - source: Data source ("gecko" or "gateway") - - Returns: - Normalized pool dict with consistent keys - """ - if source == "gecko": - # GeckoTerminal format - attrs = pool.get("attributes", pool) - - return { - "address": attrs.get("address") or pool.get("id", "").split("_")[-1], - "name": attrs.get("name", "Unknown"), - "base_token_symbol": attrs.get("base_token_symbol", "???"), - "quote_token_symbol": attrs.get("quote_token_symbol", "???"), - "base_token_price_usd": attrs.get("base_token_price_usd"), - "quote_token_price_usd": attrs.get("quote_token_price_usd"), - "network": pool.get("network") or attrs.get("network", "solana"), - "dex_id": attrs.get("dex_id", "unknown"), - "reserve_usd": attrs.get("reserve_in_usd"), - "volume_24h": _get_nested_float(attrs, "volume_usd", "h24"), - "volume_6h": _get_nested_float(attrs, "volume_usd", "h6"), - "volume_1h": _get_nested_float(attrs, "volume_usd", "h1"), - "price_change_24h": _get_nested_float( - attrs, "price_change_percentage", "h24" - ), - "price_change_6h": _get_nested_float( - attrs, "price_change_percentage", "h6" - ), - "price_change_1h": _get_nested_float( - attrs, "price_change_percentage", "h1" - ), - "fdv_usd": attrs.get("fdv_usd"), - "market_cap_usd": attrs.get("market_cap_usd"), - "pool_created_at": attrs.get("pool_created_at"), - "source": "gecko", - } - - elif source == "gateway": - # Gateway CLMM format - return { - "address": pool.get("pool_address") or pool.get("address", ""), - "name": pool.get("trading_pair") or pool.get("name", "Unknown"), - "base_token_symbol": pool.get("base_symbol", "???"), - "quote_token_symbol": pool.get("quote_symbol", "???"), - "base_token_price_usd": None, # Not provided by gateway - "quote_token_price_usd": None, - "network": "solana", - "dex_id": pool.get("connector", "meteora"), - "reserve_usd": pool.get("liquidity") or pool.get("tvl"), - "volume_24h": pool.get("volume_24h"), - "price_change_24h": None, - "current_price": pool.get("current_price") or pool.get("price"), - "bin_step": pool.get("bin_step"), - "apr": pool.get("apr"), - "apy": pool.get("apy"), - "base_fee_percentage": pool.get("base_fee_percentage"), - "mint_x": pool.get("mint_x"), - "mint_y": pool.get("mint_y"), - "source": "gateway", - } - - return pool - - -def _get_nested_float(data: dict, *keys) -> Optional[float]: - """Get a nested float value from dict, trying multiple key patterns""" - # Try nested access - value = data - for key in keys: - if isinstance(value, dict): - value = value.get(key) - else: - value = None - break - - if value is not None: - try: - return float(value) - except (ValueError, TypeError): - pass - - # Try flattened key with underscore - flat_key = "_".join(keys) - value = data.get(flat_key) - if value is not None: - try: - return float(value) - except (ValueError, TypeError): - pass - - # Try flattened key with dot - flat_key = ".".join(keys) - value = data.get(flat_key) - if value is not None: - try: - return float(value) - except (ValueError, TypeError): - pass - - return None - - -def extract_pair_from_name(name: str) -> Tuple[str, str]: - """Extract base and quote symbols from pool name - - Args: - name: Pool name like "SOL/USDC" or "SOL-USDC" or "SOL / USDC" - - Returns: - Tuple of (base_symbol, quote_symbol) - """ - if not name: - return "???", "???" - - # Try different separators - for sep in ["/", " / ", "-", " - "]: - if sep in name: - parts = name.split(sep) - if len(parts) >= 2: - return parts[0].strip(), parts[1].strip() - - return name, "???"