diff --git a/alembic/versions/003_add_daily_budget.py b/alembic/versions/003_add_daily_budget.py new file mode 100644 index 0000000..19e4006 --- /dev/null +++ b/alembic/versions/003_add_daily_budget.py @@ -0,0 +1,25 @@ +"""Add daily_budget_usd to subscriptions + +Revision ID: 003 +Revises: 002 +Create Date: 2026-04-22 00:00:00.000000 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "003" +down_revision: Union[str, None] = "002" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("subscriptions") as batch_op: + batch_op.add_column(sa.Column("daily_budget_usd", sa.Float(), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("subscriptions") as batch_op: + batch_op.drop_column("daily_budget_usd") diff --git a/alembic/versions/004_add_active_window.py b/alembic/versions/004_add_active_window.py new file mode 100644 index 0000000..fcfc15a --- /dev/null +++ b/alembic/versions/004_add_active_window.py @@ -0,0 +1,27 @@ +"""Add active_from/active_until to subscriptions + +Revision ID: 004 +Revises: 003 +Create Date: 2026-04-22 01:00:00.000000 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "004" +down_revision: Union[str, None] = "003" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("subscriptions") as batch_op: + batch_op.add_column(sa.Column("active_from", sa.DateTime(), nullable=True)) + batch_op.add_column(sa.Column("active_until", sa.DateTime(), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("subscriptions") as batch_op: + batch_op.drop_column("active_until") + batch_op.drop_column("active_from") diff --git a/alembic/versions/005_bottrade_snapshot.py b/alembic/versions/005_bottrade_snapshot.py new file mode 100644 index 0000000..85b9b63 --- /dev/null +++ b/alembic/versions/005_bottrade_snapshot.py @@ -0,0 +1,27 @@ +"""Snapshot size_usd / leverage on BotTrade + +Revision ID: 005 +Revises: 004 +Create Date: 2026-04-23 00:00:00.000000 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "005" +down_revision: Union[str, None] = "004" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("bot_trades") as batch_op: + batch_op.add_column(sa.Column("size_usd", sa.Float(), nullable=True)) + batch_op.add_column(sa.Column("leverage", sa.Integer(), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("bot_trades") as batch_op: + batch_op.drop_column("leverage") + batch_op.drop_column("size_usd") diff --git a/app/api/posts.py b/app/api/posts.py index f438841..b6549a9 100644 --- a/app/api/posts.py +++ b/app/api/posts.py @@ -26,15 +26,23 @@ def _direction_correct(signal: Optional[str], pct: Optional[float]) -> Optional[ def _post_to_schema(post: Post) -> TrumpPost: price_impact: Optional[PriceImpact] = None if post.price_impact_asset and post.price_at_post is not None: + # Overlay live rolling peaks for windows that haven't closed yet + from app.services.price_impact_monitor import get_live_impact + live = get_live_impact(post.id) or {} + + m5 = live.get("price_impact_m5", post.price_impact_m5) + m15 = live.get("price_impact_m15", post.price_impact_m15) + m1h = live.get("price_impact_m1h", post.price_impact_m1h) + price_impact = PriceImpact( asset=post.price_impact_asset, - m5=post.price_impact_m5 or 0.0, - m15=post.price_impact_m15 or 0.0, - m1h=post.price_impact_m1h or 0.0, + m5=m5, + m15=m15, + m1h=m1h, price_at_post=post.price_at_post, - correct_m5=_direction_correct(post.signal, post.price_impact_m5), - correct_m15=_direction_correct(post.signal, post.price_impact_m15), - correct_m1h=_direction_correct(post.signal, post.price_impact_m1h), + correct_m5=_direction_correct(post.signal, m5), + correct_m15=_direction_correct(post.signal, m15), + correct_m1h=_direction_correct(post.signal, m1h), ) return TrumpPost( id=post.id, diff --git a/app/api/user.py b/app/api/user.py index 30fc300..9c13962 100644 --- a/app/api/user.py +++ b/app/api/user.py @@ -1,6 +1,6 @@ import logging -from fastapi import APIRouter, Depends, Header, HTTPException, Query +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -150,6 +150,9 @@ async def get_user( take_profit_pct=sub.take_profit_pct, stop_loss_pct=sub.stop_loss_pct, min_confidence=sub.min_confidence, + daily_budget_usd=sub.daily_budget_usd, + active_from=iso_utc(sub.active_from), + active_until=iso_utc(sub.active_until), ), ) @@ -157,10 +160,20 @@ async def get_user( @router.put("/user/{wallet}/settings", response_model=UserSettings) async def set_user_settings( wallet: str, - body: SetSettingsRequest, + request: Request, db: AsyncSession = Depends(get_db), ): wallet = wallet.lower().strip() + # Read raw JSON FIRST — the frontend signs the payload exactly as it's + # serialized in JSON (e.g. `5` stays `5`, not `5.0`). If we let Pydantic + # coerce ints→floats first, the server-side hash won't match the client's. + raw = await request.json() + raw_settings = raw.get("settings") + if not isinstance(raw_settings, dict): + raise HTTPException(422, "Missing settings block") + + # Now validate with Pydantic + body = SetSettingsRequest(**raw) if wallet != body.wallet.lower().strip(): raise HTTPException(400, "Wallet mismatch") @@ -169,19 +182,46 @@ async def set_user_settings( raise HTTPException(422, "leverage must be 1–50") if not (5 <= s.position_size_usd <= 10000): raise HTTPException(422, "position_size_usd must be $5–$10,000") - if s.take_profit_pct is not None and not (0.1 <= s.take_profit_pct <= 50): - raise HTTPException(422, "take_profit_pct must be 0.1–50") - if s.stop_loss_pct is not None and not (0.1 <= s.stop_loss_pct <= 50): - raise HTTPException(422, "stop_loss_pct must be 0.1–50") + # Take-profit, stop-loss and daily budget are now MANDATORY — the bot won't + # run without them, so we reject null here and force the client to supply values. + if s.take_profit_pct is None or not (0.1 <= s.take_profit_pct <= 50): + raise HTTPException(422, "take_profit_pct is required (0.1–50)") + if s.stop_loss_pct is None or not (0.1 <= s.stop_loss_pct <= 50): + raise HTTPException(422, "stop_loss_pct is required (0.1–50)") if not (0 <= s.min_confidence <= 100): raise HTTPException(422, "min_confidence must be 0–100") + if s.daily_budget_usd is None or not (0 < s.daily_budget_usd <= 100000): + raise HTTPException(422, "daily_budget_usd is required (>0, ≤100,000)") + # Parse schedule (ISO strings → naive-UTC datetimes). Either both or neither. + from datetime import datetime as _dt, timezone as _tz + + def _parse_iso_utc(v): + if v is None: + return None + try: + # Accept "...Z" or "+00:00" forms + dt = _dt.fromisoformat(v.replace("Z", "+00:00")) + if dt.tzinfo is not None: + dt = dt.astimezone(_tz.utc).replace(tzinfo=None) + return dt + except Exception: + raise HTTPException(422, f"invalid ISO datetime: {v!r}") + + af = _parse_iso_utc(s.active_from) + au = _parse_iso_utc(s.active_until) + if (af is None) != (au is None): + raise HTTPException(422, "active_from and active_until must be set together or both empty") + if af is not None and au is not None and au <= af: + raise HTTPException(422, "active_until must be after active_from") + + # Hash the RAW dict (matches what frontend signed) verify_signed_request( action=ACTION_SET_SETTINGS, wallet=wallet, timestamp_ms=body.timestamp, signature=body.signature, - body=s.model_dump(), + body=raw_settings, ) result = await db.execute(select(Subscription).where(Subscription.wallet_address == wallet)) @@ -194,6 +234,9 @@ async def set_user_settings( sub.take_profit_pct = s.take_profit_pct sub.stop_loss_pct = s.stop_loss_pct sub.min_confidence = s.min_confidence + sub.daily_budget_usd = s.daily_budget_usd + sub.active_from = af + sub.active_until = au await db.commit() logger.info("Settings updated for %s: %s", wallet, s.model_dump()) return s diff --git a/app/main.py b/app/main.py index 8347735..aa638ff 100644 --- a/app/main.py +++ b/app/main.py @@ -10,6 +10,7 @@ from app.config import settings from app.database import AsyncSessionLocal, engine from app.models import Base from app.scrapers.truth_social import poll_truth_social, backfill_history +from app.scrapers.trumpstruth import poll_trumpstruth from app.services.binance import run_binance_ws from app.ws.manager import manager @@ -64,9 +65,24 @@ async def lifespan(app: FastAPI): max_instances=1, coalesce=True, ) + # Fallback poller — trumpstruth.org RSS. Same Truth Social originalId = + # same external_id hash, so dedup is automatic. Whoever sees the post + # first wins. Offset by half the interval so the two pollers don't + # hammer the network at the same instant. + _scheduler.add_job( + poll_trumpstruth, + "interval", + seconds=settings.truth_social_poll_seconds, + args=[AsyncSessionLocal], + id="trumpstruth_poll", + max_instances=1, + coalesce=True, + next_run_time=None, # APScheduler will pick a fresh slot + ) _scheduler.start() logger.info( - "Truth Social poller scheduled every %ds.", settings.truth_social_poll_seconds + "Truth Social pollers scheduled every %ds (CNN + trumpstruth.org).", + settings.truth_social_poll_seconds, ) yield @@ -117,8 +133,85 @@ app.include_router(user_router, prefix="/api") @app.get("/api/health") async def health(): + """Shallow liveness — used by load balancers / Docker healthcheck. + Returns 200 as long as the process is alive and the event loop responds.""" return {"status": "ok"} + +@app.get("/api/health/deep") +async def health_deep(): + """Deep healthcheck — used by external uptime monitors (UptimeRobot, etc). + + Returns 503 if any of: + - DB query fails + - Scraper hasn't completed a successful poll in > 90s + (poll runs every 15s; allow 6 misses before alarm) + + Body always includes diagnostic fields so the alarm payload is actionable. + """ + from datetime import datetime, timezone + from fastapi import Response + from sqlalchemy import text + from app.scrapers import truth_social as ts_scraper + from app.scrapers import trumpstruth as ts_fallback + + now = datetime.now(timezone.utc) + problems: list[str] = [] + + # 1. DB ping + db_ok = False + db_error: Optional[str] = None + try: + async with AsyncSessionLocal() as db: + await db.execute(text("SELECT 1")) + db_ok = True + except Exception as exc: + db_error = str(exc) + problems.append(f"db: {db_error}") + + # 2. Scraper liveness — system is healthy if AT LEAST ONE source polled + # recently. Both stale = real problem (network down, or both upstreams dead). + sources = [ + ("cnn", ts_scraper.last_successful_poll_at, ts_scraper.last_poll_error), + ("trumpstruth", ts_fallback.last_successful_poll_at, ts_fallback.last_poll_error), + ] + source_status = [] + freshest_age: Optional[int] = None + for name, last, err in sources: + age = int((now - last).total_seconds()) if last else None + source_status.append({ + "name": name, + "last_poll": last.isoformat() if last else None, + "age_sec": age, + "last_error": err, + }) + if age is not None and (freshest_age is None or age < freshest_age): + freshest_age = age + + if freshest_age is None: + problems.append("scrapers: no source has ever completed a poll") + elif freshest_age > 90: + # 6× the 15s interval — both sources are silent + problems.append(f"scrapers: all stale (freshest={freshest_age}s)") + + body = { + "status": "ok" if not problems else "degraded", + "now": now.isoformat(), + "db_ok": db_ok, + "db_error": db_error, + "scrapers": source_status, + "freshest_age_sec": freshest_age, + "problems": problems, + } + + if problems: + return Response( + content=__import__("json").dumps(body), + status_code=503, + media_type="application/json", + ) + return body + if settings.environment == "development": from app.api.dev import router as dev_router app.include_router(dev_router, prefix="/api") diff --git a/app/models.py b/app/models.py index 715e67f..03cd18e 100644 --- a/app/models.py +++ b/app/models.py @@ -90,6 +90,11 @@ class BotTrade(Base): opened_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=utcnow) closed_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) hl_order_id: Mapped[Optional[str]] = mapped_column(String(128), nullable=True) + # Snapshot of user settings AT TIME OF ENTRY. Required so changes to + # Subscription.position_size_usd / leverage after-the-fact don't corrupt + # historical PnL or daily-budget accounting. + size_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + leverage: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) trigger_post: Mapped[Optional["Post"]] = relationship("Post", back_populates="trades") @@ -110,3 +115,8 @@ class Subscription(Base): take_profit_pct: Mapped[Optional[float]] = mapped_column(Float, nullable=True) # e.g. 2.0 = close at +2% stop_loss_pct: Mapped[Optional[float]] = mapped_column(Float, nullable=True) # e.g. 1.5 = close at -1.5% min_confidence: Mapped[int] = mapped_column(Integer, nullable=False, default=80) + daily_budget_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) # e.g. 15.0 = max $15 of new positions per UTC day + # Optional active window. Both None = always on (when Subscription.active=True). + # Both stored as naive-UTC datetimes. + active_from: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) + active_until: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) diff --git a/app/schemas.py b/app/schemas.py index 7055fb4..d508931 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -5,11 +5,13 @@ from pydantic import BaseModel class PriceImpact(BaseModel): asset: str - m5: float - m15: float - m1h: float + # None = window still open (live rolling peak) or no price data yet. + # Float = sealed peak move (%) in the signal direction once window closed. + m5: Optional[float] = None + m15: Optional[float] = None + m1h: Optional[float] = None price_at_post: float - # None = outcome window not yet reached or no price data; True/False = signal direction matched actual move + # None = outcome window not yet reached; True/False = signal direction matched correct_m5: Optional[bool] = None correct_m15: Optional[bool] = None correct_m1h: Optional[bool] = None @@ -101,6 +103,10 @@ class UserSettings(BaseModel): take_profit_pct: Optional[float] = None stop_loss_pct: Optional[float] = None min_confidence: int + daily_budget_usd: Optional[float] = None + # ISO-8601 UTC strings; both None = always on (Subscription.active still gates it). + active_from: Optional[str] = None + active_until: Optional[str] = None class SetSettingsRequest(SignedEnvelope): diff --git a/app/scrapers/trumpstruth.py b/app/scrapers/trumpstruth.py new file mode 100644 index 0000000..13dde6c --- /dev/null +++ b/app/scrapers/trumpstruth.py @@ -0,0 +1,155 @@ +""" +Trump Truth Social scraper — trumpstruth.org RSS fallback. + +Why a second scraper? + CNN's archive (the primary source) sometimes lags 5–10 minutes behind real + posts. trumpstruth.org publishes an RSS feed of the same Truth Social account + that often updates faster. Running both in parallel and deduping by the + Truth Social `originalId` gives us "min(latency_a, latency_b)" — i.e. whoever + sees the post first wins. + +Dedup strategy: + Both CNN and trumpstruth expose the underlying Truth Social post id. We + hash it the same way (md5(str(id))) so the second source is a no-op when + the first already inserted the row. + +Source: https://www.trumpstruth.org/feed (RSS 2.0 with custom truth:originalId tag) +""" + +import hashlib +import logging +import re +import xml.etree.ElementTree as ET +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime +from typing import Optional + +import httpx + +from app.scrapers.truth_social import _process_entry, _post_to_ws_payload +from app.ws.manager import manager + +logger = logging.getLogger(__name__) + +FEED_URL = "https://www.trumpstruth.org/feed" +NS = { + "atom": "http://www.w3.org/2005/Atom", + "truth": "https://truthsocial.com/ns", +} + +# Liveness — read by /api/health/deep +last_successful_poll_at: Optional[datetime] = None +last_poll_error: Optional[str] = None + + +async def _fetch_feed() -> Optional[str]: + headers = { + "User-Agent": "Mozilla/5.0 (compatible; TrumpSignal/1.0)", + "Accept": "application/rss+xml, application/xml", + } + try: + async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client: + resp = await client.get(FEED_URL, headers=headers) + resp.raise_for_status() + return resp.text + except Exception as exc: + logger.warning("Failed to fetch trumpstruth.org feed: %s", exc) + return None + + +_HTML_TAG = re.compile(r"<[^>]+>") + + +def _to_cnn_shape(item: ET.Element) -> Optional[dict]: + """Convert one from the RSS feed into the dict shape the existing + `_process_entry` expects (CNN archive format). + + Required output keys: id, created_at (ISO), content (HTML).""" + orig_id_el = item.find("truth:originalId", NS) + if orig_id_el is None or not (orig_id_el.text or "").strip(): + return None + orig_id = orig_id_el.text.strip() + + pub_el = item.find("pubDate") + if pub_el is None or not pub_el.text: + return None + try: + dt = parsedate_to_datetime(pub_el.text) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + else: + dt = dt.astimezone(timezone.utc) + created_iso = dt.isoformat().replace("+00:00", "Z") + except Exception: + return None + + desc_el = item.find("description") + content_html = (desc_el.text or "").strip() if desc_el is not None else "" + + return { + "id": orig_id, + "created_at": created_iso, + "content": content_html, + } + + +async def poll_trumpstruth(db_session_factory) -> None: + """One poll cycle. Called by APScheduler. + + Idempotent: posts already inserted by the CNN scraper are skipped via the + `external_id` uniqueness check inside `_process_entry`. + """ + global last_successful_poll_at, last_poll_error + + raw = await _fetch_feed() + if raw is None: + last_poll_error = "fetch_feed returned None" + return + + try: + root = ET.fromstring(raw) + except ET.ParseError as exc: + logger.warning("trumpstruth RSS parse error: %s", exc) + last_poll_error = f"parse: {exc}" + return + + items = root.findall(".//item") + if not items: + last_poll_error = "feed had no " + return + + # Same as CNN: process the latest 50 only — keeps poll fast and avoids + # re-scanning the whole feed every 15s. + recent = items[:50] + entries = [e for e in (_to_cnn_shape(it) for it in recent) if e] + + async with db_session_factory() as db: + try: + new_posts = [] + for entry in entries: + try: + post = await _process_entry(entry, db) + if post: + new_posts.append(post) + except Exception as exc: + logger.error("trumpstruth: error on entry %s: %s", + entry.get("id"), exc) + + if new_posts: + await db.commit() + for post in new_posts: + await manager.broadcast(_post_to_ws_payload(post)) + logger.info("[trumpstruth] beat CNN — new post id=%d: %s", + post.id, post.text[:60]) + try: + from app.services.bot_engine import process_post + await process_post(post, db) + except Exception as exc: + logger.error("process_post failed for post %d: %s", + post.id, exc) + last_successful_poll_at = datetime.now(timezone.utc) + last_poll_error = None + except Exception as exc: + logger.error("trumpstruth transaction error: %s", exc) + last_poll_error = f"transaction: {exc}" + await db.rollback() diff --git a/app/scrapers/truth_social.py b/app/scrapers/truth_social.py index 380dfed..a8c4cc5 100644 --- a/app/scrapers/truth_social.py +++ b/app/scrapers/truth_social.py @@ -24,6 +24,11 @@ logger = logging.getLogger(__name__) ARCHIVE_URL = "https://ix.cnn.io/data/truth-social/truth_archive.json" +# Liveness tracker — updated every successful poll (even when no new posts). +# Read by /api/health to detect a dead scraper. None = never ran yet. +last_successful_poll_at: Optional[datetime] = None +last_poll_error: Optional[str] = None + def _strip_html(text: str) -> str: text = re.sub(r"<[^>]+>", " ", text) @@ -32,8 +37,17 @@ def _strip_html(text: str) -> str: def _parse_dt(iso: str) -> datetime: + """Parse Truth Social's ISO timestamp into a naive-UTC datetime. + + IMPORTANT: must convert to UTC *before* stripping tzinfo. Otherwise an + input like '2026-04-24T15:07:48-04:00' would be stored as naive + 15:07:48 and later mis-read as UTC — a silent 4-hour shift. + """ try: - return datetime.fromisoformat(iso.replace("Z", "+00:00")).replace(tzinfo=None) + dt = datetime.fromisoformat(iso.replace("Z", "+00:00")) + if dt.tzinfo is not None: + dt = dt.astimezone(timezone.utc) + return dt.replace(tzinfo=None) except Exception: return datetime.now(timezone.utc).replace(tzinfo=None) @@ -69,12 +83,12 @@ async def _process_entry(entry: dict, db: AsyncSession) -> Optional[Post]: analysis = await analyze_post(text) asset = analysis["asset"] - price_impact_m5 = price_impact_m15 = price_impact_m1h = price_at_post = None + # Only capture the price AT post time. The m5/m15/m1h peaks are filled in + # asynchronously by price_impact_monitor as the windows elapse — avoids + # recording 0.00% because future candles don't exist yet at entry time. + price_at_post = None if asset and analysis["relevant"]: price_at_post = price_store.get_price_at(asset, published_at) - price_impact_m5 = price_store.get_pct_change(asset, published_at, 5) - price_impact_m15 = price_store.get_pct_change(asset, published_at, 15) - price_impact_m1h = price_store.get_pct_change(asset, published_at, 60) post = Post( external_id=external_id, @@ -89,24 +103,38 @@ async def _process_entry(entry: dict, db: AsyncSession) -> Optional[Post]: analysis_version=analysis.get("analysis_version"), relevant=analysis["relevant"], price_impact_asset=asset if analysis["relevant"] else None, - price_impact_m5=price_impact_m5, - price_impact_m15=price_impact_m15, - price_impact_m1h=price_impact_m1h, + price_impact_m5=None, # filled by price_impact_monitor after 5 m + price_impact_m15=None, # filled by price_impact_monitor after 15 m + price_impact_m1h=None, # filled by price_impact_monitor after 1 h price_at_post=price_at_post, ) db.add(post) await db.flush() + + # Register with the live peak tracker so it starts watching immediately. + if asset and analysis["relevant"] and price_at_post: + from app.services.price_impact_monitor import register_post + register_post( + post_id=post.id, + asset=asset, + signal=analysis.get("signal"), + entry_price=price_at_post, + published_at=published_at, + ) + return post def _post_to_ws_payload(post: Post) -> dict: price_impact = None if post.price_impact_asset and post.price_at_post is not None: + # At broadcast time all windows are open — values are null until + # price_impact_monitor fills them in. Frontend treats null as "pending". price_impact = { "asset": post.price_impact_asset, - "m5": post.price_impact_m5 or 0.0, - "m15": post.price_impact_m15 or 0.0, - "m1h": post.price_impact_m1h or 0.0, + "m5": post.price_impact_m5, # None = not yet measured + "m15": post.price_impact_m15, + "m1h": post.price_impact_m1h, "price_at_post": post.price_at_post, } return { @@ -127,9 +155,11 @@ def _post_to_ws_payload(post: Post) -> dict: async def poll_truth_social(db_session_factory) -> None: + global last_successful_poll_at, last_poll_error logger.info("Polling CNN Truth Social archive...") entries = await _fetch_archive() if not entries: + last_poll_error = "fetch_archive returned empty" return # Only process the latest 50 entries each poll (archive has 30k+ posts) @@ -159,8 +189,12 @@ async def poll_truth_social(db_session_factory) -> None: logger.error("process_post failed for post %d: %s", post.id, exc) else: logger.info("No new posts found.") + # Mark a successful poll cycle (separate from "found new posts"). + last_successful_poll_at = datetime.now(timezone.utc) + last_poll_error = None except Exception as exc: logger.error("Transaction error: %s", exc) + last_poll_error = f"transaction_error: {exc}" await db.rollback() diff --git a/app/services/analysis.py b/app/services/analysis.py index 211123e..22a1c6e 100644 --- a/app/services/analysis.py +++ b/app/services/analysis.py @@ -207,7 +207,7 @@ async def analyze_post(text: str) -> dict: if asset not in ("BTC", "ETH", None): asset = None - reasoning = str(result.get("reasoning", ""))[:600] + reasoning = str(result.get("reasoning", ""))[:1200] return { "relevant": relevant, diff --git a/app/services/binance.py b/app/services/binance.py index dd9d209..e204faa 100644 --- a/app/services/binance.py +++ b/app/services/binance.py @@ -45,6 +45,10 @@ async def _process_message(raw: str): from app.services.tp_sl_monitor import on_price_tick on_price_tick(asset, candle["close"]) + # Price-impact peak tracker (updates rolling max for open post windows) + from app.services.price_impact_monitor import on_price_tick as pi_tick + pi_tick(asset, candle["high"], candle["low"], candle["close"]) + # Broadcast live price tick await manager.broadcast({ "type": "price", diff --git a/app/services/bot_engine.py b/app/services/bot_engine.py index 3076d1c..029bf85 100644 --- a/app/services/bot_engine.py +++ b/app/services/bot_engine.py @@ -22,15 +22,38 @@ from app.services.price_store import price_store # noqa: F401 (used elsewhere) logger = logging.getLogger(__name__) # Platform-wide thresholds (per-user values in Subscription override where applicable) -GLOBAL_MIN_CONFIDENCE = 80 # hard floor — user min_confidence must be >= this +# No global confidence floor — users pick their own 0–100 threshold via settings. MAX_HOLD_SECONDS = 3600 # force-close after 1 hour (shared across users) +# Hyperliquid perp fees (mainnet, base tier). +# IOC orders always cross the book → taker fee both on open and close. +# Ref: https://hyperliquid.gitbook.io/hyperliquid-docs/trading/fees +HL_TAKER_FEE_RATE = 0.00045 # 4.5 bps per side → 9 bps round-trip + # Per-trade locks prevent TP/SL and max-hold tasks from both calling close_and_finalize # simultaneously on the same trade. Lock is process-local — with the atomic # conditional UPDATE below, multi-process is still safe (losers become no-ops). _close_locks: Dict[int, asyncio.Lock] = {} +# Strong references to background close tasks. Python's GC can collect +# unreferenced asyncio.Task objects before they finish — keeping them here +# prevents silent task cancellation. Entries are removed when tasks complete. +_background_tasks: set = set() + +# Per-wallet locks for the open-position critical section. +# Prevents two concurrent signals from both passing the daily-budget check +# before either trade is written to DB (TOCTOU race). +_wallet_open_locks: Dict[str, asyncio.Lock] = {} + + +def _wallet_lock(wallet: str) -> asyncio.Lock: + lock = _wallet_open_locks.get(wallet) + if lock is None: + lock = asyncio.Lock() + _wallet_open_locks[wallet] = lock + return lock + def _lock_for(trade_id: int) -> asyncio.Lock: lock = _close_locks.get(trade_id) @@ -47,14 +70,12 @@ async def process_post(post: Post, db: AsyncSession) -> None: """ if not post.relevant: return - if (post.ai_confidence or 0) < GLOBAL_MIN_CONFIDENCE: - logger.info("Post %d skipped: confidence %d < %d", post.id, post.ai_confidence, GLOBAL_MIN_CONFIDENCE) - return - if post.signal not in ('buy', 'short'): + if post.signal not in ('buy', 'short', 'sell'): logger.info("Post %d skipped: signal=%s is not actionable", post.id, post.signal) return asset = post.price_impact_asset or 'BTC' + # "sell" is treated as a short signal (bearish directional trade) side = 'long' if post.signal == 'buy' else 'short' logger.info("Signal: post=%d asset=%s side=%s confidence=%d", post.id, asset, side, post.ai_confidence) @@ -76,6 +97,9 @@ async def process_post(post: Post, db: AsyncSession) -> None: take_profit_pct=s.take_profit_pct, stop_loss_pct=s.stop_loss_pct, min_confidence=s.min_confidence, + daily_budget_usd=s.daily_budget_usd, + active_from=s.active_from, + active_until=s.active_until, ) for s in subscribers ] @@ -100,70 +124,119 @@ async def _execute_for_subscriber( if not sub["hl_api_key"]: logger.warning("Subscriber %s has no HL API key, skipping", wallet) return + # Required-setup guard: the bot is only allowed to trade when the user + # has explicitly set take-profit, stop-loss, and a daily budget. Prevents + # accidental trading on partially-configured accounts. + missing = [k for k in ("take_profit_pct", "stop_loss_pct", "daily_budget_usd") + if sub.get(k) is None] + if missing: + logger.info("Sub %s skipped post %d: setup incomplete (missing %s)", + wallet, post_id, ", ".join(missing)) + return + if post_confidence < sub["min_confidence"]: logger.info("Sub %s filters out post %d: conf %d < user min %d", wallet, post_id, post_confidence, sub["min_confidence"]) return + # Active-window check. Both values stored as naive-UTC. + af = sub.get("active_from") + au = sub.get("active_until") + if af is not None and au is not None: + now_utc_naive = datetime.now(timezone.utc).replace(tzinfo=None) + if now_utc_naive < af or now_utc_naive >= au: + logger.info("Sub %s skipped post %d: outside active window [%s, %s)", + wallet, post_id, af, au) + return + try: api_key = decrypt_api_key(sub["hl_api_key"]) except Exception as exc: logger.error("Cannot decrypt key for %s: %s", wallet, exc) return - # Each subscriber gets its own session — AsyncSession is NOT concurrency-safe. - async with AsyncSessionLocal() as db: - try: - trader = HyperliquidTrader( - api_private_key=api_key, - account_address=wallet, - leverage=sub["leverage"], - mainnet=settings.hl_mainnet, - ) - - open_positions = await trader.get_open_positions() - if any(p.get('coin') == asset for p in open_positions): - logger.info("Subscriber %s already has open %s position, skipping", wallet, asset) + # Per-wallet lock wraps BOTH the budget re-check and the open, so two + # concurrent signals can't both pass the daily-budget gate before either + # trade is written to DB (TOCTOU race under asyncio.gather). + async with _wallet_lock(wallet): + # Re-check daily budget INSIDE the lock so it's atomic with the open. + daily_cap = sub.get("daily_budget_usd") + if daily_cap is not None and daily_cap > 0: + start_of_day = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=None) + async with AsyncSessionLocal() as budget_db: + spent_result = await budget_db.execute( + select(BotTrade).where( + BotTrade.wallet_address == wallet, + BotTrade.opened_at >= start_of_day, + ) + ) + spent_trades = spent_result.scalars().all() + spent = sum( + (t.size_usd if t.size_usd is not None else sub["position_size_usd"]) + for t in spent_trades + ) + if spent + sub["position_size_usd"] > daily_cap: + logger.info("Sub %s daily budget reached: spent=%.2f + new=%.2f > cap=%.2f", + wallet, spent, sub["position_size_usd"], daily_cap) return - result = await trader.open_position(asset, side, sub["position_size_usd"]) - entry_price = result.get('fill_price', 0.0) + # Each subscriber gets its own session — AsyncSession is NOT concurrency-safe. + async with AsyncSessionLocal() as db: + try: + trader = HyperliquidTrader( + api_private_key=api_key, + account_address=wallet, + leverage=sub["leverage"], + mainnet=settings.hl_mainnet, + ) - trade = BotTrade( - asset=asset, - side=side, - entry_price=entry_price, - wallet_address=wallet, - trigger_post_id=post_id, - opened_at=datetime.now(timezone.utc).replace(tzinfo=None), - hl_order_id=result.get('order_id'), - ) - db.add(trade) - await db.commit() - await db.refresh(trade) + open_positions = await trader.get_open_positions() + if any(p.get('coin') == asset for p in open_positions): + logger.info("Subscriber %s already has open %s position, skipping", wallet, asset) + return - logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)", - side, asset, wallet, entry_price, trade.id) + result = await trader.open_position(asset, side, sub["position_size_usd"]) + entry_price = result.get('fill_price', 0.0) - from app.services.tp_sl_monitor import register_trade - register_trade( - trade_id=trade.id, - wallet=wallet, - api_key=api_key, - leverage=sub["leverage"], - asset=asset, - side=side, - entry_price=entry_price, - take_profit_pct=sub["take_profit_pct"], - stop_loss_pct=sub["stop_loss_pct"], - ) + trade = BotTrade( + asset=asset, + side=side, + entry_price=entry_price, + wallet_address=wallet, + trigger_post_id=post_id, + opened_at=datetime.now(timezone.utc).replace(tzinfo=None), + hl_order_id=result.get('order_id'), + size_usd=sub["position_size_usd"], + leverage=sub["leverage"], + ) + db.add(trade) + await db.commit() + await db.refresh(trade) - asyncio.create_task(_close_after_hold( - trade.id, api_key, sub["leverage"], asset, wallet - )) + logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)", + side, asset, wallet, entry_price, trade.id) - except Exception as e: - logger.error("Trade execution failed for %s: %s", wallet, e) + from app.services.tp_sl_monitor import register_trade + register_trade( + trade_id=trade.id, + wallet=wallet, + api_key=api_key, + leverage=sub["leverage"], + asset=asset, + side=side, + entry_price=entry_price, + take_profit_pct=sub["take_profit_pct"], + stop_loss_pct=sub["stop_loss_pct"], + ) + + task = asyncio.create_task(_close_after_hold( + trade.id, api_key, sub["leverage"], asset, wallet + )) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) + + except Exception as e: + logger.error("Trade execution failed for %s: %s", wallet, e) async def close_and_finalize( @@ -200,27 +273,61 @@ async def close_and_finalize( result = await db.execute(select(BotTrade).where(BotTrade.id == trade_id)) trade = result.scalar_one() - sub_res = await db.execute( - select(Subscription).where(Subscription.wallet_address == wallet) - ) - sub = sub_res.scalar_one_or_none() - size_usd = sub.position_size_usd if sub else 20.0 + # Prefer the historical snapshot stamped on the trade row; + # fall back to current Subscription or caller's leverage for + # legacy rows. + if trade.size_usd is not None: + size_usd = trade.size_usd + else: + sub_res = await db.execute( + select(Subscription).where(Subscription.wallet_address == wallet) + ) + sub = sub_res.scalar_one_or_none() + size_usd = sub.position_size_usd if sub else 20.0 + trade_leverage = trade.leverage if trade.leverage is not None else leverage trader = HyperliquidTrader( api_private_key=api_key, account_address=wallet, - leverage=leverage, + leverage=trade_leverage, mainnet=settings.hl_mainnet, ) close_result = await trader.close_position(asset) - exit_price = close_result.get('fill_price', 0.0) + + # Detect "position was already closed externally" before we even tried + if close_result.get("already_closed"): + logger.warning( + "Trade %d: no open %s position found on HL — " + "user likely closed it manually. Marking trade closed with no PnL.", + trade_id, asset, + ) + trade.exit_price = None + trade.pnl_usd = None + trade.hold_seconds = int( + (datetime.now(timezone.utc) - + trade.opened_at.replace(tzinfo=timezone.utc)).total_seconds() + ) + await db.commit() + from app.services.tp_sl_monitor import unregister + unregister(trade_id) + _close_locks.pop(trade_id, None) + return + + exit_price = close_result.get("fill_price") + if not exit_price: + raise ValueError(f"close_position returned no fill_price for trade {trade_id}") now_aware = datetime.now(timezone.utc) opened_aware = trade.opened_at.replace(tzinfo=timezone.utc) hold_secs = int((now_aware - opened_aware).total_seconds()) pct = ((exit_price - trade.entry_price) / trade.entry_price) if trade.entry_price else 0.0 signed_pct = pct if trade.side == 'long' else -pct - pnl_usd = size_usd * signed_pct * leverage + # size_usd is the NOTIONAL value — leverage affects margin, not PnL. + gross_pnl = size_usd * signed_pct + # Deduct round-trip taker fees (IOC crosses book on both open + close). + # Without this, displayed PnL overstates real returns by ~9 bps per trade. + fees_usd = size_usd * HL_TAKER_FEE_RATE * 2 + pnl_usd = gross_pnl - fees_usd trade.exit_price = exit_price trade.pnl_usd = round(pnl_usd, 2) @@ -233,11 +340,29 @@ async def close_and_finalize( unregister(trade_id) _close_locks.pop(trade_id, None) - logger.info("Closed %s %s for %s @ %.2f PnL=%.2f (reason=%s)", - trade.side, asset, wallet, exit_price, pnl_usd, reason) + logger.info( + "Closed %s %s for %s @ %.2f gross=%.2f fees=%.2f net=%.2f (reason=%s)", + trade.side, asset, wallet, exit_price, + gross_pnl, fees_usd, pnl_usd, reason, + ) except Exception as e: logger.error("Failed to close trade %d: %s", trade_id, e) + # Rollback closed_at so recovery can retry this trade on next restart. + # Without this, the trade is "closed" in DB but position still open on HL. + try: + async with AsyncSessionLocal() as rb_db: + await rb_db.execute( + update(BotTrade) + .where(BotTrade.id == trade_id) + .values(closed_at=None) + ) + await rb_db.commit() + logger.info("Rolled back closed_at for trade %d — will retry on recovery", trade_id) + except Exception as rb_exc: + logger.error("Failed to rollback closed_at for trade %d: %s", trade_id, rb_exc) + finally: + _close_locks.pop(trade_id, None) async def _close_after_hold( diff --git a/app/services/hyperliquid.py b/app/services/hyperliquid.py index 193da05..6e394f9 100644 --- a/app/services/hyperliquid.py +++ b/app/services/hyperliquid.py @@ -111,7 +111,7 @@ class HyperliquidTrader: return positions except Exception as exc: logger.error("get_open_positions error: %s", exc) - return [] + raise async def set_leverage(self, coin: str) -> None: """Set isolated leverage for the given coin.""" @@ -180,14 +180,30 @@ class HyperliquidTrader: ) statuses = result.get("response", {}).get("data", {}).get("statuses", [{}]) - filled = statuses[0] if statuses else {} - fill_price = float( - (filled.get("filled") or {}).get("avgPx", mid) or mid - ) - order_id = str((filled.get("resting") or {}).get("oid", "")) + status = statuses[0] if statuses else {} - logger.info("Opened %s %s @ %.2f (order_id=%s)", side, coin, fill_price, order_id) - return {"order_id": order_id, "fill_price": fill_price, "size_coins": size_coins} + # "error" key present → HL rejected the order entirely + if "error" in status: + raise ValueError(f"HL order rejected: {status['error']}") + + filled_info = status.get("filled") or {} + total_sz = float(filled_info.get("totalSz", 0) or 0) + if total_sz <= 0: + # IOC returned with zero fill — no position was opened + raise ValueError( + f"IOC order for {coin} returned 0 fill " + f"(status={status}). No position opened." + ) + + fill_price = float(filled_info.get("avgPx", mid) or mid) + + # oid lives under "filled" for IOC; "resting" is fallback for unexpected GTC + oid_src = filled_info or status.get("resting") or {} + order_id = str(oid_src.get("oid", "")) if oid_src else "" + + logger.info("Opened %s %s @ %.2f size=%.5f (order_id=%s)", + side, coin, fill_price, total_sz, order_id) + return {"order_id": order_id, "fill_price": fill_price, "size_coins": total_sz} async def close_position(self, asset: str) -> dict: """ @@ -201,8 +217,9 @@ class HyperliquidTrader: positions = await self.get_open_positions() target = next((p for p in positions if p.get("coin") == coin), None) if target is None: - logger.warning("No open %s position to close", coin) - return {"fill_price": 0.0} + logger.warning("No open %s position found on HL — already closed externally?", coin) + # Return sentinel so callers can detect "nothing to close" vs "closed @ price" + return {"fill_price": None, "already_closed": True} szi = float(target["szi"]) is_buy = szi < 0 # closing a short → buy back @@ -235,10 +252,21 @@ class HyperliquidTrader: ) statuses = result.get("response", {}).get("data", {}).get("statuses", [{}]) - filled = statuses[0] if statuses else {} - fill_price = float( - (filled.get("filled") or {}).get("avgPx", mid) or mid - ) + status = statuses[0] if statuses else {} + filled_info = status.get("filled") or {} + total_sz = float(filled_info.get("totalSz", 0) or 0) - logger.info("Closed %s position @ %.2f", coin, fill_price) - return {"fill_price": fill_price} + if total_sz <= 0: + # IOC close got 0 fill — position may have already been closed externally. + # Re-check to be sure before giving up. + positions_after = await self.get_open_positions() + still_open = next((p for p in positions_after if p.get("coin") == coin), None) + if still_open is None: + logger.info("Close IOC got 0 fill but position is gone — treated as closed", ) + return {"fill_price": mid, "already_closed": False} + logger.error("Close IOC got 0 fill and position still open for %s", coin) + raise ValueError(f"Failed to close {coin} position: IOC returned 0 fill") + + fill_price = float(filled_info.get("avgPx", mid) or mid) + logger.info("Closed %s position @ %.2f (size=%.5f)", coin, fill_price, total_sz) + return {"fill_price": fill_price, "already_closed": False} diff --git a/app/services/price_impact_monitor.py b/app/services/price_impact_monitor.py new file mode 100644 index 0000000..262004d --- /dev/null +++ b/app/services/price_impact_monitor.py @@ -0,0 +1,204 @@ +""" +Rolling price-impact tracker for Trump posts. + +For each newly-saved relevant post, we track the peak favorable move +(max high for BUY signals, max low for SHORT signals) in three windows: +5 m, 15 m, and 1 h. + +Rules: + - While the window is OPEN → live rolling peak, updated on every price tick + - When the window CLOSES → final peak is written to DB, window is sealed + - When all three windows close → post is unregistered to free memory + +The result is that: + • A brand-new post immediately shows the running peak since publication. + • A post 20 minutes old shows final m5, final m15, and live 1h peak. + • A post 2 hours old shows final m5 / m15 / m1h from DB. + +Integration points: + binance.py → call on_price_tick(asset, high, low, close) each candle + truth_social.py → call register_post(...) after flush, before commit + posts.py → call get_live_impact(post_id) to overlay live values +""" + +import asyncio +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Dict, Optional + +logger = logging.getLogger(__name__) + +# Window durations in seconds +WINDOWS = {"m5": 5 * 60, "m15": 15 * 60, "m1h": 60 * 60} + + +@dataclass +class TrackedPost: + post_id: int + asset: str + signal: Optional[str] # "buy" | "short" | "sell" | "hold" | None + entry_price: float # price at post time + published_at: datetime # naive UTC + + # Running peaks — signed % relative to entry_price + # For BUY: positive = price went up (we want max) + # For SHORT: positive = price went down (we want max of -pct) + peak_m5: Optional[float] = None + peak_m15: Optional[float] = None + peak_m1h: Optional[float] = None + + # True once the window has expired and the value is finalised in DB + done_m5: bool = False + done_m15: bool = False + done_m1h: bool = False + + +# post_id → TrackedPost +_tracked: Dict[int, TrackedPost] = {} + +# Strong refs to DB-write tasks so GC doesn't collect them mid-flight +_background_tasks: set = set() + + +# ───────────────────────────────────────────────────────────────────────────── +# Public API +# ───────────────────────────────────────────────────────────────────────────── + +def register_post( + post_id: int, + asset: str, + signal: Optional[str], + entry_price: float, + published_at: datetime, +) -> None: + """Call immediately after a relevant post is flushed to DB.""" + if entry_price <= 0: + return + if not asset: + return + _tracked[post_id] = TrackedPost( + post_id=post_id, + asset=asset.upper(), + signal=signal, + entry_price=entry_price, + published_at=published_at.replace(tzinfo=None), # store as naive UTC + ) + logger.info("PriceImpact: tracking post %d (%s %s @ %.2f)", + post_id, signal, asset, entry_price) + + +def unregister(post_id: int) -> None: + _tracked.pop(post_id, None) + + +def get_live_impact(post_id: int) -> Optional[dict]: + """ + Return the current live peaks for a post if it is still being tracked. + Returns None if the post has never been registered or all windows are done. + The dict keys match the Post model columns: + price_impact_m5, price_impact_m15, price_impact_m1h + Only keys with open (not-yet-sealed) windows are included — callers + should overlay these on top of DB values. + """ + tp = _tracked.get(post_id) + if tp is None: + return None + out: dict = {} + if not tp.done_m5: + out["price_impact_m5"] = tp.peak_m5 + if not tp.done_m15: + out["price_impact_m15"] = tp.peak_m15 + if not tp.done_m1h: + out["price_impact_m1h"] = tp.peak_m1h + return out if out else None + + +def on_price_tick(asset: str, high: float, low: float, close: float) -> None: + """ + Called by binance.py on every 1-minute candle close. + Updates running peaks and fires DB-write tasks for expired windows. + """ + asset = asset.upper() + if not _tracked: + return + + now_naive = datetime.now(timezone.utc).replace(tzinfo=None) + + for post_id, tp in list(_tracked.items()): + if tp.asset != asset: + continue + + age_s = (now_naive - tp.published_at).total_seconds() + + # Pick the extreme price for this signal direction + # BUY → we care about how high price went (use candle high) + # SHORT/SELL → we care about how low price went (use candle low) + is_long = tp.signal in ("buy",) + extreme = high if is_long else low + + # signed % gain in the signal's direction + if tp.entry_price > 0: + raw_pct = (extreme - tp.entry_price) / tp.entry_price * 100 + signed_pct = raw_pct if is_long else -raw_pct + else: + signed_pct = None + + # Update each open window + for win, duration in WINDOWS.items(): + done_attr = f"done_{win}" + peak_attr = f"peak_{win}" + if getattr(tp, done_attr): + continue # already sealed + + if signed_pct is not None: + cur = getattr(tp, peak_attr) + if cur is None or signed_pct > cur: + setattr(tp, peak_attr, round(signed_pct, 4)) + + # Has the window expired? + if age_s >= duration: + setattr(tp, done_attr, True) + final_peak = getattr(tp, peak_attr) + t = asyncio.create_task(_write_window_to_db(post_id, win, final_peak)) + _background_tasks.add(t) + t.add_done_callback(_background_tasks.discard) + logger.debug("PriceImpact: post %d window %s closed → peak=%.4f%%", + post_id, win, final_peak or 0) + + # All windows done → free memory + if tp.done_m5 and tp.done_m15 and tp.done_m1h: + unregister(post_id) + logger.info("PriceImpact: post %d fully tracked, unregistered", post_id) + + +# ───────────────────────────────────────────────────────────────────────────── +# DB write helpers +# ───────────────────────────────────────────────────────────────────────────── + +_WINDOW_COLUMN = { + "m5": "price_impact_m5", + "m15": "price_impact_m15", + "m1h": "price_impact_m1h", +} + + +async def _write_window_to_db(post_id: int, window: str, value: Optional[float]) -> None: + """Write the final peak for a single window to the DB.""" + from sqlalchemy import update + from app.database import AsyncSessionLocal + from app.models import Post + + col = _WINDOW_COLUMN[window] + try: + async with AsyncSessionLocal() as db: + await db.execute( + update(Post) + .where(Post.id == post_id) + .values({col: value}) + ) + await db.commit() + logger.info("PriceImpact: wrote post %d %s=%.4f%% to DB", + post_id, window, value or 0) + except Exception as exc: + logger.error("PriceImpact: failed to write post %d %s: %s", post_id, window, exc) diff --git a/app/services/recovery.py b/app/services/recovery.py index abb64d2..2799ee1 100644 --- a/app/services/recovery.py +++ b/app/services/recovery.py @@ -18,7 +18,7 @@ logger = logging.getLogger(__name__) async def rehydrate_open_trades() -> None: # Imported locally to avoid circular imports at module load - from app.services.bot_engine import MAX_HOLD_SECONDS, _close_after_hold, close_and_finalize + from app.services.bot_engine import MAX_HOLD_SECONDS, close_and_finalize from app.services.crypto import decrypt_api_key from app.services.tp_sl_monitor import register_trade @@ -49,12 +49,16 @@ async def rehydrate_open_trades() -> None: logger.error("Cannot decrypt key for trade %d: %s", t.id, exc) continue + # Use the leverage snapshot from the trade row (stamped at open time). + # Fall back to current Subscription only for legacy rows (pre-migration 005). + trade_leverage = t.leverage if t.leverage is not None else sub.leverage + # Re-register TP/SL watcher register_trade( trade_id=t.id, wallet=t.wallet_address, api_key=api_key, - leverage=sub.leverage, + leverage=trade_leverage, asset=t.asset, side=t.side, entry_price=t.entry_price, @@ -67,23 +71,29 @@ async def rehydrate_open_trades() -> None: elapsed = (now - opened_aware).total_seconds() remaining = MAX_HOLD_SECONDS - elapsed + from app.services.bot_engine import _background_tasks + if remaining <= 0: logger.info("Trade %d past max-hold on startup — closing now", t.id) - asyncio.create_task( + task = asyncio.create_task( close_and_finalize( - trade_id=t.id, api_key=api_key, leverage=sub.leverage, + trade_id=t.id, api_key=api_key, leverage=trade_leverage, asset=t.asset, wallet=t.wallet_address, reason="max_hold_recovery", ) ) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) else: - async def _delayed_close(trade_id=t.id, key=api_key, lev=sub.leverage, + async def _delayed_close(trade_id=t.id, key=api_key, lev=trade_leverage, asset=t.asset, wallet=t.wallet_address, delay=remaining): await asyncio.sleep(delay) await close_and_finalize( trade_id=trade_id, api_key=key, leverage=lev, asset=asset, wallet=wallet, reason="max_hold", ) - asyncio.create_task(_delayed_close()) + task = asyncio.create_task(_delayed_close()) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) await db.commit() logger.info("Rehydrated %d open trades.", len(open_trades)) diff --git a/app/services/tp_sl_monitor.py b/app/services/tp_sl_monitor.py index 37c1bb0..05314f0 100644 --- a/app/services/tp_sl_monitor.py +++ b/app/services/tp_sl_monitor.py @@ -33,6 +33,9 @@ class WatchedTrade: # trade_id -> WatchedTrade _watched: Dict[int, WatchedTrade] = {} +# Strong references to fire-close tasks to prevent GC before completion +_background_tasks: set = set() + def register_trade( trade_id: int, @@ -79,7 +82,9 @@ def on_price_tick(asset: str, price: float) -> None: for wt, reason in triggered: unregister(wt.trade_id) - asyncio.create_task(_fire_close(wt, reason)) + task = asyncio.create_task(_fire_close(wt, reason)) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) async def _fire_close(wt: WatchedTrade, reason: str) -> None: diff --git a/deploy/README.md b/deploy/README.md new file mode 100644 index 0000000..90f70c4 --- /dev/null +++ b/deploy/README.md @@ -0,0 +1,155 @@ +# Backend deployment + uptime monitoring + +This folder collects the moving pieces that turn the backend from "runs on +my laptop" into "runs 24/7 with auto-restart and external alerting". + +## What the bot needs to be reliable + +1. **Auto-restart on crash** — handled by systemd (`Restart=always`) or + Docker Compose (`restart: unless-stopped`). Pick one. +2. **Health endpoints** — exposed by `app/main.py`: + - `GET /api/health` (shallow): 200 as long as the FastAPI process is alive. + Used by Docker's internal `HEALTHCHECK`. + - `GET /api/health/deep`: 200 only if DB ping works AND at least one + Truth Social scraper has polled in the last 90s. Used by external + uptime monitors. +3. **Dual-source scraper** — `truth_social.py` (CNN archive) + + `trumpstruth.py` (RSS). Both insert into `posts` with the same + `external_id` hash, so duplicates are dropped. Whoever sees the post + first wins. +4. **External uptime monitor** — pings the deep health endpoint from + outside the box. If we lose internet or the box is on fire, we get + alerted. Recommended: UptimeRobot (free tier covers this). + +--- + +## Option A — Docker Compose (recommended) + +`docker-compose.yml` is already configured with: + +```yaml +restart: unless-stopped +healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8000/api/health"] + interval: 15s + timeout: 5s + retries: 5 + start_period: 30s +``` + +Deploy steps on a fresh server: + +```bash +git clone +cd backend +cp .env.example .env +# Fill in: SERVICE_USER_POSTGRES, SERVICE_PASSWORD_POSTGRES, FRONTEND_URL, +# ENCRYPTION_KEY, AI_API_KEY, HL_MAINNET=true +docker compose up -d +docker compose logs -f api # tail logs +``` + +To verify auto-restart works: + +```bash +docker compose kill api # simulate crash +sleep 12 # wait for compose to relaunch +curl http://localhost:8000/api/health # should return {"status":"ok"} +``` + +--- + +## Option B — systemd (bare-metal, no Docker) + +Use `trumpsignal-api.service` from this folder. + +```bash +# 1. Code lives at /opt/trumpsignal/backend +sudo useradd -m -d /opt/trumpsignal trumpsignal +sudo -u trumpsignal git clone /opt/trumpsignal/backend +cd /opt/trumpsignal/backend +sudo -u trumpsignal python3.11 -m venv venv +sudo -u trumpsignal venv/bin/pip install -r requirements.txt + +# 2. Configure +sudo -u trumpsignal cp .env.example .env +sudo -u trumpsignal nano .env # fill in secrets + +# 3. Install + start the service +sudo cp deploy/trumpsignal-api.service /etc/systemd/system/ +sudo systemctl daemon-reload +sudo systemctl enable --now trumpsignal-api + +# 4. Tail logs +sudo journalctl -u trumpsignal-api -f +``` + +Test auto-restart: + +```bash +sudo systemctl kill --signal=SIGKILL trumpsignal-api +sleep 12 +curl http://localhost:8000/api/health # should return {"status":"ok"} +``` + +--- + +## UptimeRobot setup + +1. Sign up at https://uptimerobot.com (free tier = 50 monitors, 5-min interval). +2. Add new monitor: + - **Type:** HTTP(s) + - **URL:** `https://api.trump-signal.bitnews.day/api/health/deep` + (replace with your real domain) + - **Monitoring interval:** 5 min + - **Alert contacts:** add your email and/or Telegram +3. UptimeRobot will alert you when the deep healthcheck returns 503 — i.e. + when both scrapers have been silent for >90s, OR when the DB is + unreachable. + +### Why deep, not shallow + +The shallow `/api/health` returns 200 even if the scraper has died but the +FastAPI process is still serving HTTP. A monitor pointed at the shallow +endpoint would happily report "all green" while the bot silently misses +posts. Always point external monitors at `/api/health/deep`. + +--- + +## Reading the deep healthcheck + +```json +{ + "status": "ok", + "now": "2026-04-25T07:35:01.123456+00:00", + "db_ok": true, + "db_error": null, + "scrapers": [ + { "name": "cnn", "last_poll": "2026-04-25T07:34:58Z", "age_sec": 3, "last_error": null }, + { "name": "trumpstruth", "last_poll": "2026-04-25T07:34:52Z", "age_sec": 9, "last_error": null } + ], + "freshest_age_sec": 3, + "problems": [] +} +``` + +`status: "degraded"` + HTTP 503 means at least one item in `problems` is +non-empty. The most common failure modes: + +- `db: ` — Postgres/SQLite can't be reached. Check `DATABASE_URL`. +- `scrapers: all stale (freshest=Ns)` — both upstream feeds silent. Check + network egress, then check if both `ix.cnn.io` and `trumpstruth.org` are + responding from the box: `curl -sI https://ix.cnn.io/data/truth-social/truth_archive.json` + +--- + +## Burst-protection note (systemd) + +The unit file caps restarts at 5 within 60s, so a hard misconfiguration +(missing env var, etc.) stops the boot loop instead of thrashing forever. +After fixing the underlying issue: + +```bash +sudo systemctl reset-failed trumpsignal-api +sudo systemctl start trumpsignal-api +``` diff --git a/deploy/trumpsignal-api.service b/deploy/trumpsignal-api.service new file mode 100644 index 0000000..5a0f087 --- /dev/null +++ b/deploy/trumpsignal-api.service @@ -0,0 +1,37 @@ +[Unit] +Description=TrumpSignal API + Truth Social poller +After=network-online.target postgresql.service +Wants=network-online.target + +[Service] +Type=simple +User=trumpsignal +Group=trumpsignal +WorkingDirectory=/opt/trumpsignal/backend +EnvironmentFile=/opt/trumpsignal/backend/.env +ExecStart=/opt/trumpsignal/backend/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 + +# Auto-restart on any exit (crash, OOM, segfault, manual kill). +# RestartSec=10 = wait 10s between restart attempts so we don't hammer a broken state. +Restart=always +RestartSec=10 + +# Burst protection: if we crash > 5 times in 60s, give up (likely a real config bug, +# not transient). Reset by `systemctl reset-failed trumpsignal-api`. +StartLimitIntervalSec=60 +StartLimitBurst=5 + +# Hardening +NoNewPrivileges=yes +ProtectSystem=strict +ProtectHome=yes +ReadWritePaths=/opt/trumpsignal/backend +PrivateTmp=yes + +# Logging — journald captures stdout/stderr; view with `journalctl -u trumpsignal-api -f` +StandardOutput=journal +StandardError=journal +SyslogIdentifier=trumpsignal-api + +[Install] +WantedBy=multi-user.target diff --git a/trumpsignal.db b/trumpsignal.db index 55edc36..eff04d2 100644 Binary files a/trumpsignal.db and b/trumpsignal.db differ