hanlap/test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Trading Portal - FastAPI + React + PostgreSQL

Browse Source
This commit is contained in:
Han Lap
2026-07-03 13:08:09 +00:00
commit 34a1e91541
198 changed files with 35110 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/tasks/__init__.py
Executable
View File
backend/app/tasks/candle_fetcher.py
Executable
+398
View File
@@ -0,0 +1,398 @@
"""APScheduler background task that periodically fetches recent candles."""
from __future__ import annotations
import asyncio
import logging
from collections.abc import Awaitable, Callable
from datetime import datetime, timezone
from typing import Any
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI
from sqlalchemy import and_, select
from sqlalchemy.exc import InterfaceError
from sqlalchemy.orm import joinedload, selectinload
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import async_session_factory
from app.exchange.factory import factory as exchange_factory
from app.exchange.types import CandleData
from app.models.candle import Candle
from app.models.exchange import Exchange
from app.models.symbol import Symbol
from app.services.candle_service import candle_cache
logger = logging.getLogger(__name__)
# ── Rate-limit CCXT API calls (max 250 concurrent fetches) ──
_FETCH_SEMAPHORE = asyncio.Semaphore(250)
# ---------------------------------------------------------------------------
# After-fetch callbacks
# ---------------------------------------------------------------------------
# External services (e.g. WebSocket push) can register callbacks that are
# invoked after candles are successfully persisted to the database.
# Signature: async callback(exchange: str, symbol: str, timeframe: str, candle_data: dict)
_after_fetch_callbacks: list[
Callable[[str, str, str, dict[str, Any]], Awaitable[None]]
] = []
def register_after_fetch_callback(
callback: Callable[[str, str, str, dict[str, Any]], Awaitable[None]],
) -> None:
"""Register an async callback invoked after candles are saved.
The callback receives ``(exchange, symbol, timeframe, candle_data_dict)``
for every candle that was newly inserted or already existed (ON CONFLICT
DO NOTHING). Multiple callbacks are supported.
"""
_after_fetch_callbacks.append(callback)
logger.debug("Registered after-fetch callback: %s", callback.__name__)
# ---------------------------------------------------------------------------
# Timeframes we care about
# ---------------------------------------------------------------------------
_TIMEFRAMES_1M: list[str] = ["1m"]
_TIMEFRAMES_5M_PLUS: list[str] = ["5m", "15m", "30m", "1h", "4h", "1d", "1w", "1M"]
_TIMEFRAMES_OPTIMIZED: list[str] = ["15m", "30m", "1h", "4h", "1d", "1w", "1M"]
# Fetch top 100 trading symbols across ALL exchanges (is_trading flag)
# ~476 total symbols (100 bases × ~5 exchanges), batch=25 → ~100 min/cycle
# 7 timeframes × 25 symbols = 175 API calls/batch → well under semaphore=250
# Cache key prefix: "{exchange_name}:{symbol}:" — we append the timeframe later
def _cache_key_prefix(exchange_name: str, symbol: str) -> str:
return f"{exchange_name}:{symbol}:"
def _invalidate_candle_cache(exchange_name: str, symbol: str) -> None:
"""Remove all cached candle entries for a given exchange+symbol pair.
Builds exact keys for all known timeframes instead of scanning the
entire cache (which was O(cache_size × symbols) and CPU-heavy).
"""
prefix = _cache_key_prefix(exchange_name, symbol)
for tf in _TIMEFRAMES_1M + _TIMEFRAMES_5M_PLUS:
key = prefix + tf
candle_cache.pop(key, None)
# ====================================================================
# Core fetch function
# ====================================================================
async def fetch_recent_candles(
app: FastAPI,
fetch_limit: int = 2,
timeframes: list[str] | None = None,
max_symbols: int = 25, # 25 symbols/batch × 4TF = 100 API calls per 5-min tick
) -> None:
"""Fetch the latest candle(s) for active trading symbols from the DB.
Processes a batch of up to ``max_symbols`` per call, cycling through
symbols alphabetically so all trading pairs are eventually covered.
Uses is_trading=true flag — top 100 bases × ~5 exchanges ≈ 476 symbols.
Full cycle: ~100 minutes at 5-min interval.
"""
if timeframes is None:
timeframes = _TIMEFRAMES_OPTIMIZED # 15m, 1h, 4h, 1d
# Persist an offset counter via module-level list (mutable singleton)
# so the next call picks up where the last one left off.
if not hasattr(fetch_recent_candles, "_offset"):
fetch_recent_candles._offset = 0
# ── Step 1: Query symbols in a SHORT-lived session ──
# We must NOT hold the session during CCXT API calls (30-60s)
# because idle_in_transaction_session_timeout=60s kills idle connections.
async with async_session_factory() as query_db:
try:
query = (
select(Symbol)
.options(joinedload(Symbol.exchange))
.join(Exchange, Exchange.id == Symbol.exchange_id)
.where(
and_(
Symbol.is_trading == True, # noqa: E712
Symbol.is_active == True, # noqa: E712
Exchange.is_active == True, # noqa: E712
)
)
.order_by(Symbol.symbol)
)
result = await query_db.execute(query)
all_symbols: list[Symbol] = list(result.scalars().all())
except InterfaceError:
logger.warning("Symbol query failed — connection closed, skipping batch")
return
if not all_symbols:
logger.debug("No active trading symbols found — skipping candle fetch")
return
# Slice the batch using a rolling offset
total = len(all_symbols)
offset = fetch_recent_candles._offset
batch = all_symbols[offset:offset + max_symbols]
# Update / wrap the offset
fetch_recent_candles._offset = (offset + max_symbols) % total
logger.info(
"Fetching candles for %d/%d symbols (offset=%d, batch=%d-%d)",
len(batch), total, offset, offset + 1, offset + len(batch),
)
# ── Step 2: Fetch candles from CCXT WITHOUT holding DB session ──
exchange_map: dict[str, list[Symbol]] = {}
for sym in batch:
exchange_name = sym.exchange.name
exchange_map.setdefault(exchange_name, []).append(sym)
all_candle_values: list[dict[str, Any]] = []
new_candle_events: list[tuple[str, str, str, dict[str, Any]]] = []
for exchange_name, sym_list in exchange_map.items():
try:
adapter = exchange_factory.create(exchange_name)
except ValueError:
logger.warning("Unknown exchange %s — skipping", exchange_name)
continue
async def _fetch_one(sym: Symbol, tf: str):
async with _FETCH_SEMAPHORE:
try:
candles = await adapter.fetch_ohlcv(
symbol=sym.symbol,
timeframe=tf,
limit=fetch_limit,
)
return sym, tf, candles
except asyncio.CancelledError:
raise
except Exception as e:
err_str = str(e)
if "does not have market symbol" in err_str or "BadSymbol" in err_str:
logger.warning(
"Symbol %s not found on %s — marking inactive",
sym.symbol, exchange_name,
)
sym.is_active = False
else:
logger.exception(
"Failed to fetch %s %s on %s",
tf, sym.symbol, exchange_name,
)
return sym, tf, []
tasks = [_fetch_one(sym, tf) for sym in sym_list for tf in timeframes]
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, BaseException):
continue
sym, tf, candles = result
for c in candles:
all_candle_values.append(
{
"symbol_id": sym.id,
"timeframe": c.timeframe,
"timestamp": c.timestamp,
"open": c.open,
"high": c.high,
"low": c.low,
"close": c.close,
"volume": c.volume,
}
)
new_candle_events.append(
(
exchange_name,
sym.symbol,
c.timeframe,
{
"symbol": sym.symbol,
"exchange": exchange_name,
"timeframe": c.timeframe,
"timestamp": c.timestamp,
"open": c.open,
"high": c.high,
"low": c.low,
"close": c.close,
"volume": c.volume,
},
)
)
await asyncio.sleep(0.05)
# ── Step 3: Save candles in a FRESH, short-lived DB session ──
if all_candle_values:
async with async_session_factory() as save_db:
try:
from sqlalchemy.dialects.postgresql import insert as pg_insert
stmt = pg_insert(Candle).values(all_candle_values)
stmt = stmt.on_conflict_do_nothing(
index_elements=["symbol_id", "timeframe", "timestamp"]
)
await save_db.execute(stmt)
await save_db.commit()
logger.info(
"Fetched and stored %d recent candles across %d symbols (timeframes=%s)",
len(all_candle_values),
len(batch),
timeframes,
)
except InterfaceError:
logger.warning(
"Candle save: InterfaceError — connection already closed, skipping. "
"Data will be re-fetched next cycle."
)
except Exception:
logger.exception("Candle save: DB step failed")
try:
await save_db.rollback()
except InterfaceError:
pass
else:
logger.debug("No candle data fetched for this batch")
# 🔑 Session closed here — connection released back to pool!
# Non-DB operations below run without holding a pool connection.
if new_candle_events:
# --- Invoke after-fetch callbacks ---
if _after_fetch_callbacks:
for exchange_name, symbol_str, tf, candle_dict in new_candle_events:
for cb in _after_fetch_callbacks:
try:
await cb(exchange_name, symbol_str, tf, candle_dict)
except Exception:
logger.exception(
"After-fetch callback %s failed for %s:%s:%s",
cb.__name__,
exchange_name,
symbol_str,
tf,
)
# --- Batch signal analysis: one analysis per unique (exchange, symbol) ---
# Only analyze the best timeframe for trading (1h) to avoid thrashing
# when multiple TFs of the same symbol run concurrently.
processed_pairs: set[tuple[str, str]] = set()
for exchange_name, symbol_str, tf, _ in new_candle_events:
if tf != "1h":
continue # only 1h triggers trade signals
pair = (exchange_name, symbol_str)
if pair in processed_pairs:
continue
processed_pairs.add(pair)
if processed_pairs:
from app.services.signal_service import analyse_and_generate_signals
# 🔧 Optimized: semaphore 3 (was 8) — lower concurrency = lower CPU
# spike + less trade open/evict thrashing from concurrent symbol analysis.
_SIGNAL_SEMAPHORE = asyncio.Semaphore(3)
async def _analyse_one(ex_name: str, sym: str) -> None:
async with _SIGNAL_SEMAPHORE:
try:
await analyse_and_generate_signals(ex_name, sym, "1h")
except Exception:
logger.exception(
"Batch signal analysis failed for %s:%s",
ex_name, sym,
)
await asyncio.gather(
*(_analyse_one(*pair) for pair in processed_pairs),
return_exceptions=True,
)
logger.debug(
"Batch signal analysis: %d unique pairs processed",
len(processed_pairs),
)
# --- Invalidate cache ---
for sym in batch:
_invalidate_candle_cache(sym.exchange.name, sym.symbol)
# ====================================================================
# Scheduler setup
# ====================================================================
def setup_candle_scheduler(app: FastAPI) -> AsyncIOScheduler:
"""Create and configure an APScheduler ``AsyncIOScheduler``.
Jobs added:
- **1m timeframes**: ``fetch_recent_candles`` every 60 seconds.
- **5m+ timeframes**: ``fetch_recent_candles`` every 5 minutes.
The scheduler is started when the FastAPI application starts and
shut down when it stops (via the *lifespan* context manager).
"""
scheduler = AsyncIOScheduler()
# ─────────────────────────────────────────────────────────────────────
# DISABLED: 1m candles — user agreed not to fetch 1m (too heavy on DB)
# Kept as commented code for future reference.
# ─────────────────────────────────────────────────────────────────────
# scheduler.add_job(
# fetch_recent_candles,
# trigger="interval",
# seconds=120,
# args=[app, 2, _TIMEFRAMES_1M, 100],
# id="fetch_candles_1m",
# replace_existing=True,
# coalesce=True,
# max_instances=1,
# misfire_grace_time=300,
# name="Fetch 1m candles",
# )
# Every 5 minutes — 7 timeframes {15m,30m,1h,4h,1d,1w,1M}, 25 symbols/batch
# ~476 trading symbols (100 bases × ~5 exchanges) ÷ 25/batch × 5 min = ~100 min full cycle
# 7 TFs × 25 symbols = 175 API calls/batch → ~35 calls/min average (well under 250 semaphore)
scheduler.add_job(
fetch_recent_candles,
trigger="interval",
seconds=300,
args=[app, 2, _TIMEFRAMES_OPTIMIZED, 25],
id="fetch_candles_optimized",
replace_existing=True,
coalesce=True,
max_instances=1,
misfire_grace_time=600,
name="Fetch trading candles (top 100 bases, 5 exchanges, 7TFs: 15m,30m,1h,4h,1d,1w,1M)",
)
logger.info("Candle scheduler configured: top 100 bases, 5 exchanges, 7TFs, 25/batch, 5min")
# NOTE: Scheduler lifecycle is managed by main.py's lifespan handler.
# The deprecated @app.on_event() decorators do NOT fire when
# lifespan= is used in the FastAPI constructor, so we removed them.
# Call scheduler.start() in your lifespan startup block instead.
return scheduler
async def force_full_sync(app: FastAPI) -> None:
"""Run a one-time full historical candle sync on startup.
Fetches up to 500 candles per symbol/timeframe to backfill
missing data after an outage or initial deployment.
"""
logger.info("Starting one-time full candle sync (fetch_limit=500)...")
try:
await fetch_recent_candles(app, fetch_limit=500)
logger.info("One-time full candle sync completed")
except Exception:
logger.exception("One-time full candle sync failed (non-fatal)")
backend/app/tasks/exchange_sync.py
Executable
+126
View File
@@ -0,0 +1,126 @@
"""Symbol synchronisation tasks for keeping exchange market data up-to-date."""
from __future__ import annotations
import logging
from fastapi import FastAPI
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.database import async_session_factory
from app.exchange.factory import factory as exchange_factory
from app.models.exchange import Exchange
from app.models.symbol import Symbol
logger = logging.getLogger(__name__)
async def sync_exchange_symbols(
db: AsyncSession,
exchange_name: str,
) -> int:
"""Fetch all trading symbols from an exchange and upsert them into the DB.
Steps:
1. Look up the ``Exchange`` record by *exchange_name*.
2. Create an exchange adapter via ``ExchangeFactory``.
3. Call ``adapter.fetch_symbols()`` to retrieve all available symbols.
4. Upsert each symbol into the ``symbols`` table.
5. Return the total number of symbols upserted.
Raises
------
ValueError
If the exchange is unknown to the factory.
"""
# --- 1. Get exchange ---
result = await db.execute(
select(Exchange).where(Exchange.name == exchange_name)
)
exchange = result.scalar_one_or_none()
if exchange is None:
raise ValueError(
f"Exchange {exchange_name!r} not found in the database. "
"Create an Exchange record first."
)
# --- 2. Create adapter ---
adapter = exchange_factory.create(exchange_name)
# --- 3. Fetch symbols ---
symbols = await adapter.fetch_symbols()
if not symbols:
logger.warning("No symbols returned from %s", exchange_name)
return 0
# --- 4. Upsert symbols ---
values = [
{
"exchange_id": exchange.id,
"symbol": sym.symbol,
"base": sym.base,
"quote": sym.quote,
"is_active": sym.is_active,
}
for sym in symbols
]
stmt = pg_insert(Symbol).values(values)
stmt = stmt.on_conflict_do_update(
index_elements=["exchange_id", "symbol"],
set_={
"base": stmt.excluded.base,
"quote": stmt.excluded.quote,
"is_active": stmt.excluded.is_active,
},
)
await db.execute(stmt)
await db.commit()
logger.info(
"Synced %d symbols from %s",
len(values),
exchange_name,
)
return len(values)
async def sync_all_exchanges(app: FastAPI) -> dict[str, int]:
"""Sync symbols for every active exchange registered in the database.
Returns a dictionary mapping ``exchange_name`` to the number of
symbols synced.
"""
results: dict[str, int] = {}
async with async_session_factory() as db:
try:
exch_result = await db.execute(
select(Exchange).where(Exchange.is_active == True) # noqa: E712
)
exchanges: list[Exchange] = list(exch_result.scalars().all())
if not exchanges:
logger.info("No active exchanges found — nothing to sync")
return results
for exchange in exchanges:
try:
count = await sync_exchange_symbols(db, exchange.name)
results[exchange.name] = count
except Exception:
logger.exception(
"Failed to sync symbols for %s",
exchange.name,
)
results[exchange.name] = -1
return results
except Exception:
logger.exception("sync_all_exchanges failed")
await db.rollback()
return results
backend/app/tasks/stale_data_monitor.py
Executable
+163
View File
@@ -0,0 +1,163 @@
"""Stale data detection for ingested candle data."""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from fastapi import FastAPI
from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import async_session_factory
from app.models.candle import Candle
from app.models.exchange import Exchange
from app.models.symbol import Symbol
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Timeframe string → seconds mapping
# ---------------------------------------------------------------------------
_TIMEFRAME_SECONDS: dict[str, int] = {
"1m": 60,
"3m": 180,
"5m": 300,
"15m": 900,
"30m": 1800,
"1h": 3600,
"2h": 7200,
"4h": 14400,
"6h": 21600,
"8h": 28800,
"12h": 43200,
"1d": 86400,
"3d": 259200,
"1w": 604800,
"1M": 2_592_000,
}
def _timeframe_to_seconds(tf: str) -> int:
"""Convert a timeframe string to seconds.
Falls back to 3600 (1 hour) for unknown timeframes.
"""
return _TIMEFRAME_SECONDS.get(tf, 3600)
# ====================================================================
# Public API
# ====================================================================
async def check_stale_candles(
db: AsyncSession,
app: FastAPI, # noqa: ARG001 — kept for consistent signature with other tasks
) -> list[dict]:
"""Scan all active symbols and detect stale candle data.
A candle is considered stale when the timestamp of the latest candle
plus *twice* the candle timeframe duration is still in the past
(i.e. ``latest_timestamp + 2 * timeframe_seconds < now``).
Returns
-------
list[dict]
Each entry contains:
- ``symbol`` — the trading pair (e.g. ``"BTC/USDT"``)
- ``exchange`` — the exchange name
- ``timeframe`` — the candle interval
- ``last_timestamp`` — the most recent candle's timestamp (ISO-8601)
- ``staleness_minutes`` — how many minutes behind expected
"""
stale_entries: list[dict] = []
try:
# --- Get all active symbols with their exchange info ---
result = await db.execute(
select(Symbol)
.join(Exchange, Exchange.id == Symbol.exchange_id)
.where(
and_(
Symbol.is_active == True, # noqa: E712
Exchange.is_active == True, # noqa: E712
)
)
)
symbols: list[Symbol] = list(result.scalars().all())
if not symbols:
logger.debug("No active symbols found — skipping stale check")
return []
now = datetime.now(tz=timezone.utc)
# Define the timeframes to check
timeframes_to_check = ["1m", "5m", "15m", "30m", "1h", "4h", "1d"]
for db_symbol in symbols:
for tf in timeframes_to_check:
tf_seconds = _timeframe_to_seconds(tf)
stale_threshold_seconds = 2 * tf_seconds
# Get the latest candle timestamp for this symbol + timeframe
ts_result = await db.execute(
select(func.max(Candle.timestamp)).where(
and_(
Candle.symbol_id == db_symbol.id,
Candle.timeframe == tf,
)
)
)
latest_ts: datetime | None = ts_result.scalar()
if latest_ts is None:
# No data at all — flag as stale
stale_entries.append(
{
"symbol": db_symbol.symbol,
"exchange": db_symbol.exchange.name,
"timeframe": tf,
"last_timestamp": None,
"staleness_minutes": None,
"reason": "no_data",
}
)
continue
# Ensure timezone-awareness
if latest_ts.tzinfo is None:
latest_ts = latest_ts.replace(tzinfo=timezone.utc)
# Expected latest timestamp
expected_latest = latest_ts.replace(tzinfo=timezone.utc) + (
__import__("datetime").timedelta(seconds=stale_threshold_seconds)
)
if expected_latest < now:
staleness_mins = (now - latest_ts).total_seconds() / 60.0
stale_entries.append(
{
"symbol": db_symbol.symbol,
"exchange": db_symbol.exchange.name,
"timeframe": tf,
"last_timestamp": latest_ts.isoformat(),
"staleness_minutes": round(staleness_mins, 1),
"reason": "stale",
}
)
if stale_entries:
logger.warning(
"Found %d stale candle entries across %d symbols",
len(stale_entries),
len(symbols),
)
else:
logger.info("All candles are up-to-date — no stale entries detected")
return stale_entries
except Exception:
logger.exception("check_stale_candles failed")
return stale_entries