From 9a72566753915184edeccda24011e138b7569759 Mon Sep 17 00:00:00 2001 From: k Date: Mon, 20 Apr 2026 23:05:59 +0800 Subject: [PATCH] first commit --- .env.example | 4 + .gitignore | 24 ++++ README.md | 0 alembic.ini | 39 ++++++ alembic/env.py | 67 +++++++++++ alembic/versions/001_initial.py | 114 ++++++++++++++++++ app/__init__.py | 0 app/api/__init__.py | 0 app/api/dev.py | 65 ++++++++++ app/api/performance.py | 69 +++++++++++ app/api/posts.py | 63 ++++++++++ app/api/prices.py | 72 ++++++++++++ app/api/subscribe.py | 66 +++++++++++ app/api/trades.py | 46 ++++++++ app/api/user.py | 59 ++++++++++ app/config.py | 21 ++++ app/database.py | 33 ++++++ app/main.py | 151 ++++++++++++++++++++++++ app/models.py | 91 ++++++++++++++ app/schemas.py | 77 ++++++++++++ app/scrapers/__init__.py | 0 app/scrapers/truth_social.py | 202 ++++++++++++++++++++++++++++++++ app/services/__init__.py | 0 app/services/analysis.py | 125 ++++++++++++++++++++ app/services/binance.py | 102 ++++++++++++++++ app/services/bot_engine.py | 131 +++++++++++++++++++++ app/services/hyperliquid.py | 176 ++++++++++++++++++++++++++++ app/services/price_backfill.py | 144 +++++++++++++++++++++++ app/services/price_store.py | 138 ++++++++++++++++++++++ app/ws/__init__.py | 0 app/ws/manager.py | 39 ++++++ requirements.txt | 16 +++ run.py | 4 + 33 files changed, 2138 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 alembic.ini create mode 100644 alembic/env.py create mode 100644 alembic/versions/001_initial.py create mode 100644 app/__init__.py create mode 100644 app/api/__init__.py create mode 100644 app/api/dev.py create mode 100644 app/api/performance.py create mode 100644 app/api/posts.py create mode 100644 app/api/prices.py create mode 100644 app/api/subscribe.py create mode 100644 app/api/trades.py create mode 100644 app/api/user.py create mode 100644 app/config.py create mode 100644 app/database.py create mode 100644 app/main.py create mode 100644 app/models.py create mode 100644 app/schemas.py create mode 100644 app/scrapers/__init__.py create mode 100644 app/scrapers/truth_social.py create mode 100644 app/services/__init__.py create mode 100644 app/services/analysis.py create mode 100644 app/services/binance.py create mode 100644 app/services/bot_engine.py create mode 100644 app/services/hyperliquid.py create mode 100644 app/services/price_backfill.py create mode 100644 app/services/price_store.py create mode 100644 app/ws/__init__.py create mode 100644 app/ws/manager.py create mode 100644 requirements.txt create mode 100644 run.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..3043b8a --- /dev/null +++ b/.env.example @@ -0,0 +1,4 @@ +DATABASE_URL=postgresql+asyncpg://postgres:password@localhost:5432/trumpsignal +REDIS_URL=redis://localhost:6379 +ANTHROPIC_API_KEY=sk-ant-... +FRONTEND_URL=http://localhost:3001 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4770923 --- /dev/null +++ b/.gitignore @@ -0,0 +1,24 @@ +# 虚拟环境 +venv/ +env/ +.env + +# 敏感信息及配置文件 +.env +*.pem + +# 数据库 +*.db +*.sqlite +*.sqlite3 + +# Python 缓存 +__pycache__/ +*.py[cod] +*$py.class + +# macOS 系统文件 +.DS_Store + +# Alembic 产生的临时迁移记录 +# 注意：alembic/versions 中的迁移脚本应该提交，但 alembic 目录下的其他临时文件可能需要忽略 diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..7c3db3c --- /dev/null +++ b/alembic.ini @@ -0,0 +1,39 @@ +[alembic] +script_location = alembic +prepend_sys_path = . +version_path_separator = os +sqlalchemy.url = postgresql+asyncpg://postgres:password@localhost:5432/trumpsignal + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..08359d1 --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,67 @@ +import asyncio +import os +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config + +# Alembic Config object, which provides access to the values within alembic.ini +config = context.config + +# Interpret the config file for Python logging. +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +# Override sqlalchemy.url from environment variable if set +db_url = os.environ.get("DATABASE_URL") +if db_url: + config.set_main_option("sqlalchemy.url", db_url) + +# Import target metadata from the application models +from app.database import Base # noqa: E402 +import app.models # noqa: E402,F401 — ensure all models are registered + +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode (no DB connection needed).""" + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection: Connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + """Run migrations in 'online' mode using an async engine.""" + connectable = async_engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + await connectable.dispose() + + +def run_migrations_online() -> None: + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/versions/001_initial.py b/alembic/versions/001_initial.py new file mode 100644 index 0000000..d4623f5 --- /dev/null +++ b/alembic/versions/001_initial.py @@ -0,0 +1,114 @@ +"""Initial schema: posts, candles, bot_trades, subscriptions + +Revision ID: 001 +Revises: +Create Date: 2026-04-20 00:00:00.000000 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "001" +down_revision: Union[str, None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ── posts ───────────────────────────────────────────────────────────── + op.create_table( + "posts", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("external_id", sa.String(64), nullable=False), + sa.Column("text", sa.Text(), nullable=False), + sa.Column("source", sa.String(32), nullable=False, server_default="truth"), + sa.Column("published_at", sa.DateTime(), nullable=False), + sa.Column("sentiment", sa.String(16), nullable=False, server_default="neutral"), + sa.Column("ai_confidence", sa.Integer(), nullable=False, server_default="0"), + sa.Column("relevant", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("price_impact_asset", sa.String(8), nullable=True), + sa.Column("price_impact_m5", sa.Float(), nullable=True), + sa.Column("price_impact_m15", sa.Float(), nullable=True), + sa.Column("price_impact_m1h", sa.Float(), nullable=True), + sa.Column("price_at_post", sa.Float(), nullable=True), + sa.Column( + "created_at", + sa.DateTime(), + nullable=False, + server_default=sa.text("NOW()"), + ), + ) + op.create_index("ix_posts_external_id", "posts", ["external_id"], unique=True) + op.create_index("ix_posts_id", "posts", ["id"]) + + # ── candles ─────────────────────────────────────────────────────────── + op.create_table( + "candles", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("asset", sa.String(8), nullable=False), + sa.Column("timeframe", sa.String(4), nullable=False), + sa.Column("time", sa.BigInteger(), nullable=False), + sa.Column("open", sa.Float(), nullable=False), + sa.Column("high", sa.Float(), nullable=False), + sa.Column("low", sa.Float(), nullable=False), + sa.Column("close", sa.Float(), nullable=False), + sa.Column("volume", sa.Float(), nullable=False), + sa.UniqueConstraint("asset", "timeframe", "time", name="uq_candle_asset_tf_time"), + ) + op.create_index("ix_candles_asset", "candles", ["asset"]) + op.create_index("ix_candles_id", "candles", ["id"]) + op.create_index("ix_candles_time", "candles", ["time"]) + + # ── bot_trades ──────────────────────────────────────────────────────── + op.create_table( + "bot_trades", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("asset", sa.String(8), nullable=False), + sa.Column("side", sa.String(8), nullable=False), + sa.Column("entry_price", sa.Float(), nullable=False), + sa.Column("exit_price", sa.Float(), nullable=True), + sa.Column("pnl_usd", sa.Float(), nullable=True), + sa.Column("hold_seconds", sa.Integer(), nullable=True), + sa.Column("trigger_post_id", sa.Integer(), sa.ForeignKey("posts.id"), nullable=True), + sa.Column("wallet_address", sa.String(64), nullable=False), + sa.Column( + "opened_at", + sa.DateTime(), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.Column("closed_at", sa.DateTime(), nullable=True), + sa.Column("hl_order_id", sa.String(128), nullable=True), + ) + op.create_index("ix_bot_trades_id", "bot_trades", ["id"]) + op.create_index("ix_bot_trades_trigger_post_id", "bot_trades", ["trigger_post_id"]) + op.create_index("ix_bot_trades_wallet_address", "bot_trades", ["wallet_address"]) + + # ── subscriptions ───────────────────────────────────────────────────── + op.create_table( + "subscriptions", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("wallet_address", sa.String(64), nullable=False), + sa.Column("hl_api_key", sa.String(256), nullable=True), + sa.Column("active", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("subscribed_at", sa.DateTime(), nullable=True), + sa.Column( + "created_at", + sa.DateTime(), + nullable=False, + server_default=sa.text("NOW()"), + ), + sa.UniqueConstraint("wallet_address", name="uq_subscriptions_wallet_address"), + ) + op.create_index("ix_subscriptions_id", "subscriptions", ["id"]) + op.create_index( + "ix_subscriptions_wallet_address", "subscriptions", ["wallet_address"], unique=True + ) + + +def downgrade() -> None: + op.drop_table("subscriptions") + op.drop_table("bot_trades") + op.drop_table("candles") + op.drop_table("posts") diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/dev.py b/app/api/dev.py new file mode 100644 index 0000000..f035f41 --- /dev/null +++ b/app/api/dev.py @@ -0,0 +1,65 @@ +""" +Dev-only endpoint: manually insert a fake post to test the full pipeline. +""" + +import asyncio +from datetime import datetime, timezone +from fastapi import APIRouter, BackgroundTasks, Depends +from sqlalchemy.ext.asyncio import AsyncSession +from app.database import get_db, AsyncSessionLocal +from app.models import Post +from app.ws.manager import manager +import hashlib, time + +router = APIRouter() + + +@router.post("/dev/fake-post") +async def insert_fake_post( + text: str = "BITCOIN is the future of money! We will make America the greatest crypto capital EVER!", + sentiment: str = "bullish", + db: AsyncSession = Depends(get_db), +): + external_id = hashlib.md5(f"fake-{time.time()}".encode()).hexdigest() + post = Post( + external_id=external_id, + text=text, + source="truth", + published_at=datetime.now(timezone.utc).replace(tzinfo=None), + sentiment=sentiment, + ai_confidence=88, + relevant=True, + price_impact_asset="BTC", + price_impact_m5=1.2, + price_impact_m15=2.4, + price_impact_m1h=3.1, + price_at_post=94230.0, + ) + db.add(post) + await db.commit() + await db.refresh(post) + + await manager.broadcast({ + "type": "new_post", + "post": { + "id": post.id, + "text": post.text, + "source": post.source, + "published_at": post.published_at.isoformat(), + "sentiment": post.sentiment, + "ai_confidence": post.ai_confidence, + "relevant": post.relevant, + "price_impact": { + "asset": "BTC", "m5": 1.2, "m15": 2.4, "m1h": 3.1, "price_at_post": 94230.0 + }, + }, + }) + return {"id": post.id, "text": post.text[:80]} + + +@router.post("/dev/backfill-prices") +async def trigger_price_backfill(background_tasks: BackgroundTasks, asset: str = "BTC"): + """触发历史帖子价格回溯（后台运行）""" + from app.services.price_backfill import backfill_price_impact + background_tasks.add_task(backfill_price_impact, AsyncSessionLocal, asset) + return {"status": "started", "asset": asset, "message": "后台回溯中，约需 1-2 分钟"} diff --git a/app/api/performance.py b/app/api/performance.py new file mode 100644 index 0000000..8be51fd --- /dev/null +++ b/app/api/performance.py @@ -0,0 +1,69 @@ +import logging +from datetime import datetime, timedelta, timezone + +from fastapi import APIRouter, Depends +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models import BotTrade +from app.schemas import BotPerformance + +router = APIRouter() +logger = logging.getLogger(__name__) + +PERIOD_DAYS = 30 + + +@router.get("/performance", response_model=BotPerformance) +async def get_performance(db: AsyncSession = Depends(get_db)): + since = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=PERIOD_DAYS) + + result = await db.execute( + select(BotTrade) + .where(BotTrade.closed_at.is_not(None)) + .where(BotTrade.opened_at >= since) + .order_by(BotTrade.opened_at.asc()) + ) + trades = result.scalars().all() + + total_trades = len(trades) + if total_trades == 0: + return BotPerformance( + period_days=PERIOD_DAYS, + total_trades=0, + win_rate=0.0, + net_pnl_usd=0.0, + avg_hold_seconds=0.0, + max_drawdown_pct=0.0, + ) + + winning = sum(1 for t in trades if (t.pnl_usd or 0) > 0) + win_rate = winning / total_trades + + pnl_values = [(t.pnl_usd or 0.0) for t in trades] + net_pnl = sum(pnl_values) + + hold_values = [(t.hold_seconds or 0) for t in trades] + avg_hold = sum(hold_values) / len(hold_values) + + # Max drawdown: running peak → trough of cumulative PnL + cumulative = 0.0 + peak = 0.0 + max_drawdown = 0.0 + for pnl in pnl_values: + cumulative += pnl + if cumulative > peak: + peak = cumulative + drawdown = (peak - cumulative) / peak * 100 if peak > 0 else 0.0 + if drawdown > max_drawdown: + max_drawdown = drawdown + + return BotPerformance( + period_days=PERIOD_DAYS, + total_trades=total_trades, + win_rate=round(win_rate, 4), + net_pnl_usd=round(net_pnl, 2), + avg_hold_seconds=round(avg_hold, 1), + max_drawdown_pct=round(max_drawdown, 4), + ) diff --git a/app/api/posts.py b/app/api/posts.py new file mode 100644 index 0000000..9e97c75 --- /dev/null +++ b/app/api/posts.py @@ -0,0 +1,63 @@ +import logging +from typing import List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models import Post +from app.schemas import PriceImpact, TrumpPost + +router = APIRouter() +logger = logging.getLogger(__name__) + + +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 + ): + 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, + price_at_post=post.price_at_post, + ) + return TrumpPost( + id=post.id, + text=post.text, + source=post.source, + published_at=post.published_at.isoformat(), + sentiment=post.sentiment, + signal=post.signal, + ai_confidence=post.ai_confidence, + ai_reasoning=post.ai_reasoning, + relevant=post.relevant, + price_impact=price_impact, + ) + + +@router.get("/posts", response_model=List[TrumpPost]) +async def get_posts( + limit: int = Query(default=20, ge=1, le=500), + page: int = Query(default=1, ge=1), + db: AsyncSession = Depends(get_db), +): + offset = (page - 1) * limit + result = await db.execute( + select(Post).order_by(Post.published_at.desc()).offset(offset).limit(limit) + ) + posts = result.scalars().all() + return [_post_to_schema(p) for p in posts] + + +@router.get("/posts/{post_id}", response_model=TrumpPost) +async def get_post(post_id: int, db: AsyncSession = Depends(get_db)): + result = await db.execute(select(Post).where(Post.id == post_id)) + post = result.scalar_one_or_none() + if post is None: + raise HTTPException(status_code=404, detail="Post not found") + return _post_to_schema(post) diff --git a/app/api/prices.py b/app/api/prices.py new file mode 100644 index 0000000..5782b10 --- /dev/null +++ b/app/api/prices.py @@ -0,0 +1,72 @@ +import logging +from typing import List + +import httpx +from fastapi import APIRouter, HTTPException, Query + +from app.config import settings +from app.schemas import Candle +from app.services.price_store import price_store + +router = APIRouter() +logger = logging.getLogger(__name__) + +VALID_ASSETS = {"BTC", "ETH"} +VALID_TIMEFRAMES = {"5m", "15m", "1H", "4H", "1D", "1W"} + +BINANCE_INTERVAL = { + "5m": "5m", + "15m": "15m", + "1H": "1h", + "4H": "4h", + "1D": "1d", + "1W": "1w", +} + +SYMBOL_MAP = {"BTC": "BTCUSDT", "ETH": "ETHUSDT"} + + +async def fetch_binance_candles(asset: str, tf: str, limit: int) -> List[Candle]: + symbol = SYMBOL_MAP[asset] + interval = BINANCE_INTERVAL[tf] + url = f"{settings.binance_rest_url}/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}" + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.get(url) + resp.raise_for_status() + rows = resp.json() + return [ + Candle( + time=row[0] // 1000, # ms → seconds + open=float(row[1]), + high=float(row[2]), + low=float(row[3]), + close=float(row[4]), + volume=float(row[5]), + ) + for row in rows + ] + + +@router.get("/prices/{asset}", response_model=List[Candle]) +async def get_prices( + asset: str, + tf: str = Query(default="4H"), + limit: int = Query(default=200, ge=1, le=1000), +): + asset = asset.upper() + if asset not in VALID_ASSETS: + raise HTTPException(status_code=400, detail=f"Asset must be one of {VALID_ASSETS}") + if tf not in VALID_TIMEFRAMES: + raise HTTPException(status_code=400, detail=f"Timeframe must be one of {VALID_TIMEFRAMES}") + + # 1m: use in-memory store (updated in real-time) + if tf == "1m": + candles = price_store.get_candles(asset, "1m", limit) + return [Candle(**c) for c in candles] + + # All other timeframes: fetch directly from Binance + try: + return await fetch_binance_candles(asset, tf, limit) + except Exception as exc: + logger.error("Binance REST fetch failed for %s %s: %s", asset, tf, exc) + raise HTTPException(status_code=502, detail="Failed to fetch price data") diff --git a/app/api/subscribe.py b/app/api/subscribe.py new file mode 100644 index 0000000..a562929 --- /dev/null +++ b/app/api/subscribe.py @@ -0,0 +1,66 @@ +import logging +from datetime import datetime, timezone + +from eth_account import Account +from eth_account.messages import encode_defunct +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models import Subscription +from app.schemas import SubscribeRequest, SubscribeResponse + +router = APIRouter() +logger = logging.getLogger(__name__) + +SIGN_MESSAGE = "Sign in to TrumpSignal" + + +from typing import Optional + +def _recover_signer(signature: str) -> Optional[str]: + """Recover signer address from an EIP-191 personal_sign signature.""" + try: + message = encode_defunct(text=SIGN_MESSAGE) + recovered = Account.recover_message(message, signature=signature) + return recovered.lower() + except Exception as exc: + logger.warning("Signature recovery failed: %s", exc) + return None + + +@router.post("/subscribe", response_model=SubscribeResponse) +async def subscribe(body: SubscribeRequest, db: AsyncSession = Depends(get_db)): + wallet = body.wallet.lower().strip() + + # Verify EIP-191 signature + recovered = _recover_signer(body.signature) + if recovered is None or recovered != wallet: + raise HTTPException( + status_code=401, + detail="Signature verification failed: recovered address does not match wallet", + ) + + # Upsert subscription + result = await db.execute( + select(Subscription).where(Subscription.wallet_address == wallet) + ) + sub = result.scalar_one_or_none() + now = datetime.now(timezone.utc).replace(tzinfo=None) + + if sub is None: + sub = Subscription( + wallet_address=wallet, + active=True, + subscribed_at=now, + ) + db.add(sub) + else: + sub.active = True + if sub.subscribed_at is None: + sub.subscribed_at = now + + await db.commit() + logger.info("Subscription activated for wallet %s", wallet) + return SubscribeResponse(status="ok", wallet=wallet) diff --git a/app/api/trades.py b/app/api/trades.py new file mode 100644 index 0000000..c58b1e8 --- /dev/null +++ b/app/api/trades.py @@ -0,0 +1,46 @@ +import logging +from typing import List + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models import BotTrade +from app.schemas import BotTrade as BotTradeSchema + +router = APIRouter() +logger = logging.getLogger(__name__) + + +def _trade_to_schema(trade: BotTrade) -> BotTradeSchema: + return BotTradeSchema( + id=trade.id, + asset=trade.asset, + side=trade.side, + entry_price=trade.entry_price, + exit_price=trade.exit_price or 0.0, + pnl_usd=trade.pnl_usd or 0.0, + hold_seconds=trade.hold_seconds or 0, + trigger_post_id=trade.trigger_post_id or 0, + opened_at=trade.opened_at.isoformat(), + closed_at=trade.closed_at.isoformat() if trade.closed_at else "", + ) + + +@router.get("/trades", response_model=List[BotTradeSchema]) +async def get_trades( + limit: int = Query(default=20, ge=1, le=100), + page: int = Query(default=1, ge=1), + db: AsyncSession = Depends(get_db), +): + offset = (page - 1) * limit + result = await db.execute( + select(BotTrade) + .where(BotTrade.closed_at.is_not(None)) + .order_by(BotTrade.opened_at.desc()) + .offset(offset) + .limit(limit) + ) + trades = result.scalars().all() + return [_trade_to_schema(t) for t in trades] diff --git a/app/api/user.py b/app/api/user.py new file mode 100644 index 0000000..dee7b4f --- /dev/null +++ b/app/api/user.py @@ -0,0 +1,59 @@ +import logging + +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models import BotTrade, Subscription +from app.schemas import BotTrade as BotTradeSchema, UserResponse + +router = APIRouter() +logger = logging.getLogger(__name__) + + +def _trade_to_schema(trade: BotTrade) -> BotTradeSchema: + return BotTradeSchema( + id=trade.id, + asset=trade.asset, + side=trade.side, + entry_price=trade.entry_price, + exit_price=trade.exit_price or 0.0, + pnl_usd=trade.pnl_usd or 0.0, + hold_seconds=trade.hold_seconds or 0, + trigger_post_id=trade.trigger_post_id or 0, + opened_at=trade.opened_at.isoformat(), + closed_at=trade.closed_at.isoformat() if trade.closed_at else "", + ) + + +@router.get("/user/{wallet}", response_model=UserResponse) +async def get_user(wallet: str, db: AsyncSession = Depends(get_db)): + wallet = wallet.lower().strip() + + result = await db.execute( + select(Subscription).where(Subscription.wallet_address == wallet) + ) + sub = result.scalar_one_or_none() + if sub is None: + raise HTTPException(status_code=404, detail="Wallet not found") + + # Personal trades (all — open and closed) + trades_result = await db.execute( + select(BotTrade) + .where(BotTrade.wallet_address == wallet) + .order_by(BotTrade.opened_at.desc()) + .limit(50) + ) + trades = trades_result.scalars().all() + + # Mask hl_api_key: show only last 4 chars if set + hl_api_key_set = bool(sub.hl_api_key) + + return UserResponse( + wallet_address=sub.wallet_address, + active=sub.active, + subscribed_at=sub.subscribed_at.isoformat() if sub.subscribed_at else None, + hl_api_key_set=hl_api_key_set, + trades=[_trade_to_schema(t) for t in trades], + ) diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..cf73802 --- /dev/null +++ b/app/config.py @@ -0,0 +1,21 @@ +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + database_url: str + redis_url: str = "redis://localhost:6379" + anthropic_api_key: str + frontend_url: str = "http://localhost:3001" + truth_social_rss: str = "https://truthsocial.com/@realDonaldTrump.rss" + truth_social_poll_seconds: int = 15 + binance_ws_url: str = ( + "wss://data-stream.binance.vision/stream" + "?streams=btcusdt@kline_1m/ethusdt@kline_1m" + ) + binance_rest_url: str = "https://data-api.binance.vision" + environment: str = "development" + + model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} + + +settings = Settings() diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..75f37c3 --- /dev/null +++ b/app/database.py @@ -0,0 +1,33 @@ +from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker +from sqlalchemy.orm import declarative_base + +from app.config import settings + +_url = settings.database_url +# SQLite for local dev — no pool settings +if _url.startswith("sqlite"): + engine = create_async_engine(_url, echo=False, connect_args={"check_same_thread": False}) +else: + engine = create_async_engine(_url, echo=False, pool_pre_ping=True, pool_size=10, max_overflow=20) + +AsyncSessionLocal = async_sessionmaker( + bind=engine, + class_=AsyncSession, + expire_on_commit=False, + autocommit=False, + autoflush=False, +) + +Base = declarative_base() + + +async def get_db(): + async with AsyncSessionLocal() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + finally: + await session.close() diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..cbb757e --- /dev/null +++ b/app/main.py @@ -0,0 +1,151 @@ +import asyncio +import logging +from contextlib import asynccontextmanager + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from fastapi import FastAPI, WebSocket, WebSocketDisconnect +from fastapi.middleware.cors import CORSMiddleware + +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.services.binance import run_binance_ws +from app.ws.manager import manager + +# Import all routers +from app.api.posts import router as posts_router +from app.api.prices import router as prices_router +from app.api.trades import router as trades_router +from app.api.performance import router as performance_router +from app.api.subscribe import router as subscribe_router +from app.api.user import router as user_router + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", +) +logger = logging.getLogger(__name__) + +from typing import Optional +_binance_task: Optional[asyncio.Task] = None +_scheduler: Optional[AsyncIOScheduler] = None + + +@asynccontextmanager +async def lifespan(app: FastAPI): + global _binance_task, _scheduler + + # 1. Create DB tables (dev convenience; production uses Alembic) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + logger.info("Database tables ensured.") + + # 2. Backfill historical posts on startup (fast, no Claude API call) + asyncio.create_task(backfill_history(AsyncSessionLocal, limit=500)) + + # 3. Start Binance WebSocket background task + _binance_task = asyncio.create_task(run_binance_ws(), name="binance_ws") + logger.info("Binance WebSocket task started.") + + # 3. Start Truth Social poller via APScheduler + _scheduler = AsyncIOScheduler() + _scheduler.add_job( + poll_truth_social, + "interval", + seconds=settings.truth_social_poll_seconds, + args=[AsyncSessionLocal], + id="truth_social_poll", + max_instances=1, + coalesce=True, + ) + _scheduler.start() + logger.info( + "Truth Social poller scheduled every %ds.", settings.truth_social_poll_seconds + ) + + yield + + # Shutdown + logger.info("Shutting down background tasks...") + if _scheduler: + _scheduler.shutdown(wait=False) + if _binance_task and not _binance_task.done(): + _binance_task.cancel() + try: + await _binance_task + except asyncio.CancelledError: + pass + await engine.dispose() + logger.info("Shutdown complete.") + + +app = FastAPI( + title="TrumpSignal API", + version="1.0.0", + description="Crypto trading signals derived from Trump's Truth Social posts.", + lifespan=lifespan, +) + +# CORS +allowed_origins = [ + settings.frontend_url, + "http://localhost:3001", + "http://localhost:3000", +] +app.add_middleware( + CORSMiddleware, + allow_origins=allowed_origins, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# REST routers +app.include_router(posts_router, prefix="/api") +app.include_router(prices_router, prefix="/api") +app.include_router(trades_router, prefix="/api") +app.include_router(performance_router, prefix="/api") +app.include_router(subscribe_router, prefix="/api") +app.include_router(user_router, prefix="/api") + + +@app.get("/api/health") +async def health(): + return {"status": "ok"} + +if settings.environment == "development": + from app.api.dev import router as dev_router + app.include_router(dev_router, prefix="/api") + + +# ── WebSocket endpoints ──────────────────────────────────────────────────── + + +@app.websocket("/ws/prices") +async def ws_prices(websocket: WebSocket): + """Subscribe to live price ticks (type: 'price').""" + await manager.connect(websocket) + try: + while True: + # Keep connection alive; messages are pushed via manager.broadcast + await websocket.receive_text() + except WebSocketDisconnect: + await manager.disconnect(websocket) + except Exception as exc: + logger.warning("WebSocket /ws/prices error: %s", exc) + await manager.disconnect(websocket) + + +@app.websocket("/ws/posts") +async def ws_posts(websocket: WebSocket): + """Subscribe to new post events (type: 'new_post').""" + await manager.connect(websocket) + try: + while True: + await websocket.receive_text() + except WebSocketDisconnect: + await manager.disconnect(websocket) + except Exception as exc: + logger.warning("WebSocket /ws/posts error: %s", exc) + await manager.disconnect(websocket) diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..584f8e4 --- /dev/null +++ b/app/models.py @@ -0,0 +1,91 @@ +from datetime import datetime, timezone +from typing import List, Optional + +from sqlalchemy import ( + BigInteger, + Boolean, + DateTime, + Float, + ForeignKey, + Integer, + String, + Text, + UniqueConstraint, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.database import Base + + +def utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +class Post(Base): + __tablename__ = "posts" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + external_id: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False) + text: Mapped[str] = mapped_column(Text, nullable=False) + source: Mapped[str] = mapped_column(String(32), nullable=False, default="truth") + published_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + sentiment: Mapped[str] = mapped_column(String(16), nullable=False, default="neutral") + ai_confidence: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + relevant: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + price_impact_asset: Mapped[Optional[str]] = mapped_column(String(8), nullable=True) + price_impact_m5: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + price_impact_m15: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + price_impact_m1h: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + price_at_post: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + signal: Mapped[Optional[str]] = mapped_column(String(8), nullable=True) + ai_reasoning: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=utcnow) + + trades: Mapped[List["BotTrade"]] = relationship("BotTrade", back_populates="trigger_post") + + +class Candle(Base): + __tablename__ = "candles" + __table_args__ = (UniqueConstraint("asset", "timeframe", "time", name="uq_candle_asset_tf_time"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + asset: Mapped[str] = mapped_column(String(8), nullable=False, index=True) + timeframe: Mapped[str] = mapped_column(String(4), nullable=False) + time: Mapped[int] = mapped_column(BigInteger, nullable=False, index=True) + open: Mapped[float] = mapped_column(Float, nullable=False) + high: Mapped[float] = mapped_column(Float, nullable=False) + low: Mapped[float] = mapped_column(Float, nullable=False) + close: Mapped[float] = mapped_column(Float, nullable=False) + volume: Mapped[float] = mapped_column(Float, nullable=False) + + +class BotTrade(Base): + __tablename__ = "bot_trades" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + asset: Mapped[str] = mapped_column(String(8), nullable=False) + side: Mapped[str] = mapped_column(String(8), nullable=False) + entry_price: Mapped[float] = mapped_column(Float, nullable=False) + exit_price: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + pnl_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + hold_seconds: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + trigger_post_id: Mapped[Optional[int]] = mapped_column( + Integer, ForeignKey("posts.id"), nullable=True, index=True + ) + wallet_address: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + 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) + + trigger_post: Mapped[Optional["Post"]] = relationship("Post", back_populates="trades") + + +class Subscription(Base): + __tablename__ = "subscriptions" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + wallet_address: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False) + hl_api_key: Mapped[Optional[str]] = mapped_column(String(256), nullable=True) + active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + subscribed_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=utcnow) diff --git a/app/schemas.py b/app/schemas.py new file mode 100644 index 0000000..c428877 --- /dev/null +++ b/app/schemas.py @@ -0,0 +1,77 @@ +from typing import Optional + +from pydantic import BaseModel + + +class PriceImpact(BaseModel): + asset: str + m5: float + m15: float + m1h: float + price_at_post: float + + +class TrumpPost(BaseModel): + id: int + text: str + source: str + published_at: str # ISO + sentiment: str + signal: Optional[str] = None # buy | sell | short | hold + ai_confidence: int + ai_reasoning: Optional[str] = None + relevant: bool + price_impact: Optional[PriceImpact] = None + + model_config = {"from_attributes": True} + + +class Candle(BaseModel): + time: int + open: float + high: float + low: float + close: float + volume: float + + +class BotTrade(BaseModel): + id: int + asset: str + side: str + entry_price: float + exit_price: float + pnl_usd: float + hold_seconds: int + trigger_post_id: int + opened_at: str + closed_at: str + + model_config = {"from_attributes": True} + + +class BotPerformance(BaseModel): + period_days: int + total_trades: int + win_rate: float + net_pnl_usd: float + avg_hold_seconds: float + max_drawdown_pct: float + + +class SubscribeRequest(BaseModel): + wallet: str + signature: str + + +class SubscribeResponse(BaseModel): + status: str + wallet: str + + +class UserResponse(BaseModel): + wallet_address: str + active: bool + subscribed_at: Optional[str] = None + hl_api_key_set: bool + trades: list[BotTrade] diff --git a/app/scrapers/__init__.py b/app/scrapers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/scrapers/truth_social.py b/app/scrapers/truth_social.py new file mode 100644 index 0000000..dd0a6a6 --- /dev/null +++ b/app/scrapers/truth_social.py @@ -0,0 +1,202 @@ +""" +Trump Truth Social scraper — CNN public archive +Source: https://ix.cnn.io/data/truth-social/truth_archive.json +Updated every ~5 minutes by CNN. +""" + +import hashlib +import html +import logging +import re +from datetime import datetime, timezone +from typing import Optional + +import httpx +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models import Post +from app.services.analysis import analyze_post +from app.services.price_store import price_store +from app.ws.manager import manager + +logger = logging.getLogger(__name__) + +ARCHIVE_URL = "https://ix.cnn.io/data/truth-social/truth_archive.json" + + +def _strip_html(text: str) -> str: + text = re.sub(r"<[^>]+>", " ", text) + text = html.unescape(text) + return re.sub(r"\s+", " ", text).strip() + + +def _parse_dt(iso: str) -> datetime: + try: + return datetime.fromisoformat(iso.replace("Z", "+00:00")).replace(tzinfo=None) + except Exception: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +async def _fetch_archive() -> Optional[list]: + headers = { + "User-Agent": "Mozilla/5.0 (compatible; TrumpSignal/1.0)", + "Accept": "application/json", + } + try: + async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client: + resp = await client.get(ARCHIVE_URL, headers=headers) + resp.raise_for_status() + return resp.json() + except Exception as exc: + logger.error("Failed to fetch CNN archive: %s", exc) + return None + + +async def _process_entry(entry: dict, db: AsyncSession) -> Optional[Post]: + external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest() + + result = await db.execute(select(Post).where(Post.external_id == external_id)) + if result.scalar_one_or_none(): + return None + + text = _strip_html(entry.get("content") or "").strip() + if not text: + return None + + published_at = _parse_dt(entry.get("created_at", "")) + + analysis = await analyze_post(text) + + asset = analysis["asset"] + price_impact_m5 = price_impact_m15 = price_impact_m1h = 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, + text=text, + source="truth", + published_at=published_at, + sentiment=analysis["sentiment"], + signal=analysis.get("signal"), + ai_confidence=analysis["confidence"], + ai_reasoning=analysis.get("reasoning"), + 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_at_post=price_at_post, + ) + db.add(post) + await db.flush() + 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: + 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, + "price_at_post": post.price_at_post, + } + return { + "type": "new_post", + "post": { + "id": post.id, + "text": post.text, + "source": post.source, + "published_at": post.published_at.isoformat(), + "sentiment": post.sentiment, + "ai_confidence": post.ai_confidence, + "relevant": post.relevant, + "price_impact": price_impact, + }, + } + + +async def poll_truth_social(db_session_factory) -> None: + logger.info("Polling CNN Truth Social archive...") + entries = await _fetch_archive() + if not entries: + return + + # Only process the latest 50 entries each poll (archive has 30k+ posts) + recent = entries[:50] + logger.info("Checking %d recent entries...", len(recent)) + + async with db_session_factory() as db: + try: + new_posts = [] + for entry in recent: + try: + post = await _process_entry(entry, db) + if post: + new_posts.append(post) + except Exception as exc: + logger.error("Error processing 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("Saved 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) + else: + logger.info("No new posts found.") + except Exception as exc: + logger.error("Transaction error: %s", exc) + await db.rollback() + + +async def backfill_history(db_session_factory, limit: int = 500) -> None: + """One-time backfill of historical posts (no Claude analysis, no price impact).""" + logger.info("Starting historical backfill (limit=%d)...", limit) + entries = await _fetch_archive() + if not entries: + logger.error("Backfill failed: could not fetch archive") + return + + to_process = entries[:limit] + saved = 0 + + async with db_session_factory() as db: + try: + for entry in to_process: + external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest() + result = await db.execute(select(Post).where(Post.external_id == external_id)) + if result.scalar_one_or_none(): + continue + + text = _strip_html(entry.get("content") or "").strip() + if not text: + continue + + post = Post( + external_id=external_id, + text=text, + source="truth", + published_at=_parse_dt(entry.get("created_at", "")), + sentiment="neutral", + ai_confidence=0, + relevant=False, + ) + db.add(post) + saved += 1 + + await db.commit() + logger.info("Backfill complete: saved %d posts", saved) + except Exception as exc: + logger.error("Backfill error: %s", exc) + await db.rollback() diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/analysis.py b/app/services/analysis.py new file mode 100644 index 0000000..44b3bca --- /dev/null +++ b/app/services/analysis.py @@ -0,0 +1,125 @@ +import json +import logging + +import anthropic + +from app.config import settings + +logger = logging.getLogger(__name__) + +from typing import Optional +_client: Optional[anthropic.AsyncAnthropic] = None + +SYSTEM_PROMPT = ( + "You are a crypto trading signal analyst. Analyze Donald Trump's social media posts " + "and determine their likely impact on cryptocurrency markets. " + "Return only valid JSON with no additional text or markdown." +) + +USER_PROMPT_TEMPLATE = """Analyze this Trump post for cryptocurrency trading signals. + +Post: {text} + +Respond with JSON only: +{{ + "relevant": , + "asset": "BTC" | "ETH" | null, + "sentiment": "bullish" | "bearish" | "neutral", + "signal": "buy" | "sell" | "short" | "hold", + "confidence": <0-100>, + "reasoning": "<1-2 sentences explaining the signal and why>" +}} + +Signal rules: +- "buy": bullish crypto/macro signal → expect price rise → go long +- "sell": bearish signal for existing longs → exit position +- "short": strong bearish signal → expect price drop → go short +- "hold": relevant but unclear direction, or neutral macro + +Bullish examples: pro-crypto policy, BTC reserve, anti-regulation, dollar weakness, tariff deal, market optimism +Bearish examples: SEC crackdowns, war escalation, economic sanctions, crypto ban threats, crisis +Not relevant: personal attacks, sports, endorsements, unrelated politics + +Be strict on relevance — most posts are NOT relevant to crypto.""" + +_FALLBACK = { + "relevant": False, + "asset": None, + "sentiment": "neutral", + "signal": "hold", + "confidence": 0, + "reasoning": "", +} + + +def _get_client() -> anthropic.AsyncAnthropic: + global _client + if _client is None: + _client = anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key) + return _client + + +async def analyze_post(text: str) -> dict: + """Run Claude signal analysis on a Trump post. + + Returns dict with keys: relevant, asset, sentiment, signal, confidence, reasoning. + Falls back to neutral hold on any error. + """ + try: + client = _get_client() + message = await client.messages.create( + model="claude-haiku-4-5-20251001", + max_tokens=300, + system=SYSTEM_PROMPT, + messages=[ + { + "role": "user", + "content": USER_PROMPT_TEMPLATE.format(text=text[:2000]), + } + ], + ) + raw = message.content[0].text.strip() + + if raw.startswith("```"): + lines = raw.split("\n") + raw = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:]) + + result = json.loads(raw) + + sentiment = result.get("sentiment", "neutral") + if sentiment not in ("bullish", "bearish", "neutral"): + sentiment = "neutral" + + signal = result.get("signal", "hold") + if signal not in ("buy", "sell", "short", "hold"): + signal = "hold" + + confidence = int(result.get("confidence", 0)) + confidence = max(0, min(100, confidence)) + + relevant = bool(result.get("relevant", False)) + + asset = result.get("asset") + if asset not in ("BTC", "ETH", None): + asset = None + + reasoning = str(result.get("reasoning", ""))[:500] + + return { + "relevant": relevant, + "asset": asset, + "sentiment": sentiment, + "signal": signal, + "confidence": confidence, + "reasoning": reasoning, + } + + except anthropic.APIError as exc: + logger.error("Anthropic API error: %s", exc) + return dict(_FALLBACK) + except json.JSONDecodeError as exc: + logger.error("Failed to parse Claude JSON response: %s", exc) + return dict(_FALLBACK) + except Exception as exc: + logger.error("Unexpected error in analyze_post: %s", exc) + return dict(_FALLBACK) diff --git a/app/services/binance.py b/app/services/binance.py new file mode 100644 index 0000000..7869b95 --- /dev/null +++ b/app/services/binance.py @@ -0,0 +1,102 @@ +import asyncio +import json +import logging + +import httpx +import websockets + +from app.config import settings +from app.services.price_store import price_store +from app.ws.manager import manager + +logger = logging.getLogger(__name__) + +ASSET_MAP = { + "btcusdt": "BTC", + "ethusdt": "ETH", +} + + +async def _process_message(raw: str): + try: + data = json.loads(raw) + # Combined stream wraps payload under "data" + payload = data.get("data", data) + if payload.get("e") != "kline": + return + + kline = payload["k"] + symbol = kline["s"].lower() + asset = ASSET_MAP.get(symbol) + if asset is None: + return + + candle = { + "time": kline["t"], # open time ms + "open": float(kline["o"]), + "high": float(kline["h"]), + "low": float(kline["l"]), + "close": float(kline["c"]), + "volume": float(kline["v"]), + } + price_store.update(asset, candle) + + # Broadcast live price tick + await manager.broadcast({ + "type": "price", + "asset": asset, + "price": candle["close"], + "time": candle["time"], + }) + except Exception as exc: + logger.warning("Error processing Binance message: %s", exc) + + +async def fetch_historical(asset: str, symbol: str, interval: str = "1m", limit: int = 500): + """Fetch historical klines from Binance REST API to pre-fill price_store.""" + url = f"{settings.binance_rest_url}/api/v3/klines?symbol={symbol.upper()}&interval={interval}&limit={limit}" + try: + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.get(url) + resp.raise_for_status() + for row in resp.json(): + candle = { + "time": row[0], # open time ms + "open": float(row[1]), + "high": float(row[2]), + "low": float(row[3]), + "close": float(row[4]), + "volume": float(row[5]), + } + price_store.update(asset, candle) + logger.info("Loaded %d historical %s candles for %s", limit, interval, asset) + except Exception as exc: + logger.error("Failed to fetch historical data for %s: %s", asset, exc) + + +async def run_binance_ws(): + """Connect to Binance kline stream with exponential back-off on disconnect.""" + # Pre-fill with historical data + await fetch_historical("BTC", "btcusdt", limit=500) + await fetch_historical("ETH", "ethusdt", limit=500) + + backoff = 1 + while True: + try: + logger.info("Connecting to Binance WebSocket: %s", settings.binance_ws_url) + async with websockets.connect( + settings.binance_ws_url, + ping_interval=20, + ping_timeout=10, + ) as ws: + backoff = 1 # reset on successful connection + logger.info("Binance WebSocket connected.") + async for message in ws: + await _process_message(message) + except asyncio.CancelledError: + logger.info("Binance WebSocket task cancelled.") + return + except Exception as exc: + logger.error("Binance WebSocket error: %s. Reconnecting in %ds.", exc, backoff) + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 60) diff --git a/app/services/bot_engine.py b/app/services/bot_engine.py new file mode 100644 index 0000000..8ede288 --- /dev/null +++ b/app/services/bot_engine.py @@ -0,0 +1,131 @@ +""" +Trading decision loop. +Called by the Truth Social scraper after each new post is analyzed and saved. +Iterates all active subscribers and executes trades on their behalf. +""" + +import asyncio +import logging +from datetime import datetime, timezone +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select +from app.models import Post, BotTrade, Subscription +from app.services.hyperliquid import HyperliquidTrader +from app.services.price_store import price_store + +logger = logging.getLogger(__name__) + +# Thresholds +MIN_CONFIDENCE = 70 # ai_confidence must be >= this to trade +POSITION_SIZE_USD = 100 # per-trade size in USD (fixed for MVP) +MAX_HOLD_SECONDS = 3600 # force-close after 1 hour + + +async def process_post(post: Post, db: AsyncSession) -> None: + """ + Entry point called by the scraper after a new post is saved. + Skips non-relevant or low-confidence posts. + """ + if not post.relevant: + return + if (post.ai_confidence or 0) < MIN_CONFIDENCE: + logger.info("Post %d skipped: confidence %d < %d", post.id, post.ai_confidence, MIN_CONFIDENCE) + return + if post.sentiment == 'neutral': + return + + asset = post.price_impact_asset or 'BTC' + side = 'long' if post.sentiment == 'bullish' else 'short' + + logger.info("Signal: post=%d asset=%s side=%s confidence=%d", post.id, asset, side, post.ai_confidence) + + result = await db.execute(select(Subscription).where(Subscription.active == True)) + subscribers = result.scalars().all() + + if not subscribers: + logger.info("No active subscribers, skipping trade execution") + return + + tasks = [_execute_for_subscriber(sub, post, asset, side, db) for sub in subscribers] + await asyncio.gather(*tasks, return_exceptions=True) + + +async def _execute_for_subscriber( + sub: Subscription, + post: Post, + asset: str, + side: str, + db: AsyncSession, +) -> None: + if not sub.hl_api_key: + logger.warning("Subscriber %s has no HL API key, skipping", sub.wallet_address) + return + + try: + trader = HyperliquidTrader(sub.hl_api_key) + + # Check for existing open position to avoid doubling up + open_positions = await trader.get_open_positions() + already_open = any(p.get('coin') == asset for p in open_positions) + if already_open: + logger.info("Subscriber %s already has open %s position, skipping", sub.wallet_address, asset) + return + + result = await trader.open_position(asset, side, POSITION_SIZE_USD) + entry_price = result.get('fill_price', 0.0) + + trade = BotTrade( + asset=asset, + side=side, + entry_price=entry_price, + wallet_address=sub.wallet_address, + 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) + + logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)", side, asset, sub.wallet_address, entry_price, trade.id) + + # Schedule close after MAX_HOLD_SECONDS + asyncio.create_task(_close_after_hold(trade.id, sub.hl_api_key, asset, sub.wallet_address)) + + except Exception as e: + logger.error("Trade execution failed for %s: %s", sub.wallet_address, e) + + +async def _close_after_hold(trade_id: int, api_key: str, asset: str, wallet: str) -> None: + await asyncio.sleep(MAX_HOLD_SECONDS) + + from app.database import AsyncSessionLocal + async with AsyncSessionLocal() as db: + try: + result = await db.execute( + select(BotTrade).where(BotTrade.id == trade_id, BotTrade.closed_at.is_(None)) + ) + trade = result.scalar_one_or_none() + if trade is None: + return # already closed + + trader = HyperliquidTrader(api_key) + close_result = await trader.close_position(asset) + exit_price = close_result.get('fill_price', 0.0) + + now = datetime.now(timezone.utc) + hold_secs = int((now - trade.opened_at.replace(tzinfo=timezone.utc)).total_seconds()) + pnl = (exit_price - trade.entry_price) * (1 if trade.side == 'long' else -1) + # Approximate PnL: size_usd * pct_change + pnl_usd = POSITION_SIZE_USD * (pnl / trade.entry_price) if trade.entry_price else 0.0 + + trade.exit_price = exit_price + trade.pnl_usd = round(pnl_usd, 2) + trade.hold_seconds = hold_secs + trade.closed_at = now + + await db.commit() + logger.info("Closed %s %s for %s @ %.2f PnL=%.2f", trade.side, asset, wallet, exit_price, pnl_usd) + + except Exception as e: + logger.error("Failed to close trade %d: %s", trade_id, e) diff --git a/app/services/hyperliquid.py b/app/services/hyperliquid.py new file mode 100644 index 0000000..1fe1148 --- /dev/null +++ b/app/services/hyperliquid.py @@ -0,0 +1,176 @@ +import asyncio +import logging +from functools import partial + +from eth_account import Account +from hyperliquid.exchange import Exchange +from hyperliquid.info import Info + +logger = logging.getLogger(__name__) + +HL_MAINNET = "https://api.hyperliquid.xyz" + +# Coin name mapping: our asset -> Hyperliquid coin +COIN_MAP = { + "BTC": "BTC", + "ETH": "ETH", +} + + +class HyperliquidTrader: + def __init__(self, private_key: str): + self.account = Account.from_key(private_key) + self.exchange = Exchange(self.account, HL_MAINNET) + self.info = Info(HL_MAINNET) + self._loop = asyncio.get_event_loop() + + # ── helpers ────────────────────────────────────────────────────────────── + + async def _run_sync(self, fn, *args, **kwargs): + """Run a blocking SDK call in the default thread pool.""" + return await self._loop.run_in_executor(None, partial(fn, *args, **kwargs)) + + def _wallet(self) -> str: + return self.account.address + + # ── public interface ───────────────────────────────────────────────────── + + async def get_balance(self) -> float: + """Return USDC balance (withdrawable).""" + try: + state = await self._run_sync( + self.info.user_state, self._wallet() + ) + return float(state.get("withdrawable", 0)) + except Exception as exc: + logger.error("get_balance error: %s", exc) + return 0.0 + + async def open_position(self, asset: str, side: str, size_usd: float) -> dict: + """Place a market order. + + Args: + asset: "BTC" or "ETH" + side: "long" (buy) or "short" (sell) + size_usd: notional size in USD + + Returns: + {"order_id": str, "fill_price": float} + """ + coin = COIN_MAP.get(asset.upper(), asset.upper()) + is_buy = side.lower() == "long" + + try: + # Get current price to compute size in coins + meta = await self._run_sync(self.info.meta) + universe = {u["name"]: u for u in meta.get("universe", [])} + coin_meta = universe.get(coin, {}) + sz_decimals = int(coin_meta.get("szDecimals", 3)) + + # Use mid-price approximation from user state + all_mids = await self._run_sync(self.info.all_mids) + mid_price = float(all_mids.get(coin, 0)) + if mid_price <= 0: + raise ValueError(f"Could not get mid price for {coin}") + + size_coins = round(size_usd / mid_price, sz_decimals) + + # Slippage: 1% for longs (limit above mid), 1% for shorts (limit below mid) + slippage = 0.01 + if is_buy: + limit_px = round(mid_price * (1 + slippage), 1) + else: + limit_px = round(mid_price * (1 - slippage), 1) + + order_result = await self._run_sync( + self.exchange.order, + coin, + is_buy, + size_coins, + limit_px, + {"limit": {"tif": "Ioc"}}, # IOC ~ market + ) + + status = order_result.get("response", {}).get("data", {}) + filled = status.get("statuses", [{}])[0] + fill_price = float(filled.get("filled", {}).get("avgPx", mid_price) or mid_price) + order_id = str(filled.get("resting", {}).get("oid", "")) + + logger.info( + "Opened %s %s position: size=%.4f coins @ %.2f", + side, coin, size_coins, fill_price, + ) + return {"order_id": order_id, "fill_price": fill_price} + + except Exception as exc: + logger.error("open_position error: %s", exc) + raise + + async def close_position(self, asset: str) -> dict: + """Close all open positions for the given asset. + + Returns: + {"fill_price": float} + """ + coin = COIN_MAP.get(asset.upper(), asset.upper()) + try: + 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 position found for %s", coin) + return {"fill_price": 0.0} + + szi = float(target.get("szi", 0)) + is_buy = szi < 0 # closing a short means buying + + all_mids = await self._run_sync(self.info.all_mids) + mid_price = float(all_mids.get(coin, 0)) + + slippage = 0.01 + if is_buy: + limit_px = round(mid_price * (1 + slippage), 1) + else: + limit_px = round(mid_price * (1 - slippage), 1) + + close_result = await self._run_sync( + self.exchange.order, + coin, + is_buy, + abs(szi), + limit_px, + {"limit": {"tif": "Ioc"}}, + reduce_only=True, + ) + + status = close_result.get("response", {}).get("data", {}) + filled = status.get("statuses", [{}])[0] + fill_price = float(filled.get("filled", {}).get("avgPx", mid_price) or mid_price) + + logger.info("Closed %s position @ %.2f", coin, fill_price) + return {"fill_price": fill_price} + + except Exception as exc: + logger.error("close_position error: %s", exc) + raise + + async def get_open_positions(self) -> list: + """Return list of open positions from Hyperliquid.""" + try: + state = await self._run_sync(self.info.user_state, self._wallet()) + positions = [] + for asset_pos in state.get("assetPositions", []): + pos = asset_pos.get("position", {}) + szi = float(pos.get("szi", 0)) + if szi != 0: + positions.append({ + "coin": pos.get("coin"), + "szi": szi, + "entry_px": float(pos.get("entryPx", 0) or 0), + "unrealized_pnl": float(pos.get("unrealizedPnl", 0) or 0), + }) + return positions + except Exception as exc: + logger.error("get_open_positions error: %s", exc) + return [] diff --git a/app/services/price_backfill.py b/app/services/price_backfill.py new file mode 100644 index 0000000..2a223d0 --- /dev/null +++ b/app/services/price_backfill.py @@ -0,0 +1,144 @@ +""" +历史帖子价格回溯 +对 DB 中没有价格数据的帖子，从 Binance 拉历史 K 线计算涨跌幅 +""" + +import asyncio +import logging +from datetime import datetime, timezone, timedelta +from typing import Optional + +import httpx +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.models import Post + +logger = logging.getLogger(__name__) + +# 每次从 Binance 拉多少分钟的 1m K 线（覆盖 1h 涨跌幅计算需要至少 60 根） +FETCH_WINDOW_MINUTES = 90 + + +async def _fetch_klines(symbol: str, start_ms: int, limit: int = 90) -> list: + url = ( + f"{settings.binance_rest_url}/api/v3/klines" + f"?symbol={symbol}&interval=1m&startTime={start_ms}&limit={limit}" + ) + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.get(url) + resp.raise_for_status() + return resp.json() + + +def _price_at(klines: list, target_ms: int) -> Optional[float]: + """找最接近 target_ms 的收盘价""" + best = None + best_diff = float("inf") + for row in klines: + diff = abs(row[0] - target_ms) + if diff < best_diff: + best_diff = diff + best = float(row[4]) # close + return best + + +def _pct_change(klines: list, from_ms: int, delta_minutes: int) -> Optional[float]: + from_price = _price_at(klines, from_ms) + to_price = _price_at(klines, from_ms + delta_minutes * 60 * 1000) + if from_price and to_price and from_price != 0: + return round((to_price - from_price) / from_price * 100, 4) + return None + + +async def backfill_price_impact(db_session_factory, asset: str = "BTC") -> None: + """ + 对 DB 中 relevant=True 但没有价格数据的帖子补充价格回溯。 + 对 relevant=False 的帖子，做简单判断（标题含 crypto/bitcoin/btc 关键词则标记为相关）。 + """ + symbol = "BTCUSDT" if asset == "BTC" else "ETHUSDT" + + async with db_session_factory() as db: + # 拿所有没有价格数据的帖子 + result = await db.execute( + select(Post) + .where(Post.price_at_post == None) + .order_by(Post.published_at.asc()) + ) + posts = result.scalars().all() + + logger.info("找到 %d 条帖子需要价格回溯 (asset=%s)", len(posts), asset) + if not posts: + return + + # 关键词判断是否与加密相关（快速，不用 Claude API） + crypto_keywords = [ + "bitcoin", "btc", "crypto", "cryptocurrency", "blockchain", + "ethereum", "eth", "digital currency", "defi", "coinbase", + "sec", "regulation", "tariff", "dollar", "inflation", "fed", + "economy", "market", "trade", "sanctions", "iran", "china", + ] + + def is_relevant(text: str) -> bool: + t = text.lower() + return any(kw in t for kw in crypto_keywords) + + saved = 0 + errors = 0 + + for i, post in enumerate(posts): + try: + published_at = post.published_at + if published_at.tzinfo is None: + published_at = published_at.replace(tzinfo=timezone.utc) + + start_ms = int(published_at.timestamp() * 1000) + + # 判断相关性（用关键词，不消耗 Claude API） + relevant = is_relevant(post.text) + sentiment = "neutral" + if relevant: + t = post.text.lower() + bullish_kw = ["great", "win", "winning", "strong", "best", "love", "beautiful", "tremendous", "amazing", "pro-crypto", "bitcoin reserve"] + bearish_kw = ["bad", "terrible", "war", "crisis", "sanction", "ban", "regulate", "crack", "fraud", "scam"] + if any(k in t for k in bullish_kw): + sentiment = "bullish" + elif any(k in t for k in bearish_kw): + sentiment = "bearish" + + # 拉 Binance 历史价格 + klines = await _fetch_klines(symbol, start_ms, limit=FETCH_WINDOW_MINUTES) + + price_at_post = _price_at(klines, start_ms) + m5 = _pct_change(klines, start_ms, 5) + m15 = _pct_change(klines, start_ms, 15) + m1h = _pct_change(klines, start_ms, 60) + + # 更新帖子 + async with db_session_factory() as db: + result = await db.execute(select(Post).where(Post.id == post.id)) + p = result.scalar_one_or_none() + if p: + p.relevant = relevant + p.sentiment = sentiment + p.price_impact_asset = asset if relevant else None + p.price_at_post = price_at_post + p.price_impact_m5 = m5 if relevant else None + p.price_impact_m15 = m15 if relevant else None + p.price_impact_m1h = m1h if relevant else None + await db.commit() + saved += 1 + + if (i + 1) % 20 == 0: + logger.info("进度: %d/%d 已处理，已保存 %d 条", i + 1, len(posts), saved) + + # 避免触发 Binance 限速（1200 requests/min） + await asyncio.sleep(0.1) + + except Exception as exc: + errors += 1 + logger.error("帖子 id=%d 回溯失败: %s", post.id, exc) + await asyncio.sleep(1) + + logger.info("✅ 价格回溯完成: 共处理 %d 条，成功 %d 条，失败 %d 条", len(posts), saved, errors) diff --git a/app/services/price_store.py b/app/services/price_store.py new file mode 100644 index 0000000..f66c3ad --- /dev/null +++ b/app/services/price_store.py @@ -0,0 +1,138 @@ +import logging +from collections import deque +from datetime import datetime, timezone +from typing import Dict, List, Optional + +logger = logging.getLogger(__name__) + +# 7 days of 1-minute candles +MAX_CANDLES = 7 * 24 * 60 # 10080 + +TIMEFRAME_MINUTES: Dict[str, int] = { + "1m": 1, + "1H": 60, + "4H": 240, + "1D": 1440, + "1W": 10080, +} + + +class PriceStore: + def __init__(self): + self._candles: Dict[str, deque] = { + "BTC": deque(maxlen=MAX_CANDLES), + "ETH": deque(maxlen=MAX_CANDLES), + } + + def update(self, asset: str, candle: dict): + """Add or replace the latest 1m candle for the asset.""" + asset = asset.upper() + if asset not in self._candles: + self._candles[asset] = deque(maxlen=MAX_CANDLES) + buf = self._candles[asset] + # If the last candle has the same timestamp, replace it (in-progress bar) + if buf and buf[-1]["time"] == candle["time"]: + buf[-1] = candle + else: + buf.append(candle) + + def get_price_at(self, asset: str, timestamp: datetime) -> Optional[float]: + """Return the close price of the candle closest to timestamp.""" + asset = asset.upper() + buf = self._candles.get(asset) + if not buf: + return None + ts = timestamp.replace(tzinfo=None) + target_unix = int(ts.timestamp()) * 1000 # candle times are ms + best = None + best_diff = float("inf") + for candle in buf: + diff = abs(candle["time"] - target_unix) + if diff < best_diff: + best_diff = diff + best = candle + return best["close"] if best else None + + def get_pct_change( + self, asset: str, from_ts: datetime, minutes: int + ) -> Optional[float]: + """Return % change from from_ts to from_ts + minutes.""" + asset = asset.upper() + buf = self._candles.get(asset) + if not buf: + return None + + from_unix_ms = int(from_ts.replace(tzinfo=None).timestamp()) * 1000 + to_unix_ms = from_unix_ms + minutes * 60 * 1000 + + # Find candle at from_ts + from_candle = self._closest_candle(buf, from_unix_ms) + to_candle = self._closest_candle(buf, to_unix_ms) + + if from_candle is None or to_candle is None: + return None + if from_candle["close"] == 0: + return None + + pct = (to_candle["close"] - from_candle["close"]) / from_candle["close"] * 100 + return round(pct, 4) + + def _closest_candle(self, buf: deque, target_unix_ms: int) -> Optional[dict]: + best = None + best_diff = float("inf") + for candle in buf: + diff = abs(candle["time"] - target_unix_ms) + if diff < best_diff: + best_diff = diff + best = candle + return best + + def get_candles(self, asset: str, timeframe: str = "1H", limit: int = 200) -> List[dict]: + """Aggregate 1m candles into the requested timeframe and return the last `limit` bars.""" + asset = asset.upper() + buf = self._candles.get(asset) + if not buf: + return [] + + tf_minutes = TIMEFRAME_MINUTES.get(timeframe, 60) + + if tf_minutes == 1: + candles = list(buf)[-limit:] + return [{**c, "time": c["time"] // 1000} for c in candles] + + # Aggregate + aggregated: Dict[int, dict] = {} + tf_ms = tf_minutes * 60 * 1000 + + for candle in buf: + bucket = (candle["time"] // tf_ms) * tf_ms + if bucket not in aggregated: + aggregated[bucket] = { + "time": bucket, + "open": candle["open"], + "high": candle["high"], + "low": candle["low"], + "close": candle["close"], + "volume": candle["volume"], + } + else: + agg = aggregated[bucket] + agg["high"] = max(agg["high"], candle["high"]) + agg["low"] = min(agg["low"], candle["low"]) + agg["close"] = candle["close"] + agg["volume"] += candle["volume"] + + sorted_candles = sorted(aggregated.values(), key=lambda c: c["time"]) + result = sorted_candles[-limit:] + # Convert ms → seconds for lightweight-charts + return [{**c, "time": c["time"] // 1000} for c in result] + + def latest_price(self, asset: str) -> Optional[float]: + asset = asset.upper() + buf = self._candles.get(asset) + if not buf: + return None + return buf[-1]["close"] + + +price_store = PriceStore() diff --git a/app/ws/__init__.py b/app/ws/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/ws/manager.py b/app/ws/manager.py new file mode 100644 index 0000000..5071bd9 --- /dev/null +++ b/app/ws/manager.py @@ -0,0 +1,39 @@ +import json +import logging +from typing import List + +from fastapi import WebSocket + +logger = logging.getLogger(__name__) + + +class ConnectionManager: + def __init__(self): + self.active_connections: List[WebSocket] = [] + + async def connect(self, ws: WebSocket): + await ws.accept() + self.active_connections.append(ws) + logger.info("WebSocket connected. Total connections: %d", len(self.active_connections)) + + async def disconnect(self, ws: WebSocket): + if ws in self.active_connections: + self.active_connections.remove(ws) + logger.info("WebSocket disconnected. Total connections: %d", len(self.active_connections)) + + async def broadcast(self, message: dict): + if not self.active_connections: + return + payload = json.dumps(message) + dead: List[WebSocket] = [] + for connection in self.active_connections: + try: + await connection.send_text(payload) + except Exception as exc: + logger.warning("Failed to send to WebSocket: %s", exc) + dead.append(connection) + for ws in dead: + await self.disconnect(ws) + + +manager = ConnectionManager() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3adac3c --- /dev/null +++ b/requirements.txt @@ -0,0 +1,16 @@ +fastapi==0.111.0 +uvicorn[standard]==0.29.0 +sqlalchemy[asyncio]==2.0.30 +asyncpg==0.29.0 +alembic==1.13.1 +pydantic==2.7.1 +pydantic-settings==2.2.1 +httpx==0.27.0 +feedparser==6.0.11 +websockets==12.0 +anthropic==0.28.0 +hyperliquid-python-sdk==0.8.0 +eth-account==0.11.0 +redis==5.0.4 +apscheduler==3.10.4 +python-dotenv==1.0.1 diff --git a/run.py b/run.py new file mode 100644 index 0000000..55bbf84 --- /dev/null +++ b/run.py @@ -0,0 +1,4 @@ +import uvicorn + +if __name__ == "__main__": + uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)