k
121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
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, iso_utc
|
|
from app.schemas import PriceImpact, TrumpPost
|
|
|
|
router = APIRouter()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _direction_correct(signal: Optional[str], pct: Optional[float]) -> Optional[bool]:
|
|
if pct is None or signal is None:
|
|
return None
|
|
if signal == "buy":
|
|
return pct > 0
|
|
if signal in ("short", "sell"):
|
|
return pct < 0
|
|
return None # hold has no direction
|
|
|
|
|
|
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,
|
|
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),
|
|
)
|
|
return TrumpPost(
|
|
id=post.id,
|
|
text=post.text,
|
|
source=post.source,
|
|
published_at=iso_utc(post.published_at),
|
|
sentiment=post.sentiment,
|
|
signal=post.signal,
|
|
ai_confidence=post.ai_confidence,
|
|
ai_reasoning=post.ai_reasoning,
|
|
prefilter_reason=post.prefilter_reason,
|
|
analysis_version=post.analysis_version,
|
|
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("/signals/accuracy")
|
|
async def signal_accuracy(db: AsyncSession = Depends(get_db)):
|
|
"""Aggregate accuracy of directional signals (buy/sell/short) against realised price moves."""
|
|
result = await db.execute(
|
|
select(Post).where(Post.signal.in_(["buy", "sell", "short"]))
|
|
)
|
|
posts = result.scalars().all()
|
|
|
|
def bucket():
|
|
return {"checked": 0, "correct": 0}
|
|
|
|
stats = {"m5": bucket(), "m15": bucket(), "m1h": bucket()}
|
|
by_signal: dict[str, dict] = {}
|
|
|
|
for p in posts:
|
|
sig = p.signal
|
|
if sig not in by_signal:
|
|
by_signal[sig] = {"m5": bucket(), "m15": bucket(), "m1h": bucket(), "count": 0}
|
|
by_signal[sig]["count"] += 1
|
|
for win, val in (("m5", p.price_impact_m5), ("m15", p.price_impact_m15), ("m1h", p.price_impact_m1h)):
|
|
ok = _direction_correct(sig, val)
|
|
if ok is None:
|
|
continue
|
|
stats[win]["checked"] += 1
|
|
stats[win]["correct"] += int(ok)
|
|
by_signal[sig][win]["checked"] += 1
|
|
by_signal[sig][win]["correct"] += int(ok)
|
|
|
|
def pct(b): return round(b["correct"] / b["checked"] * 100, 1) if b["checked"] else None
|
|
|
|
return {
|
|
"overall": {k: {**v, "accuracy_pct": pct(v)} for k, v in stats.items()},
|
|
"by_signal": {
|
|
s: {
|
|
"count": d["count"],
|
|
"m5": {**d["m5"], "accuracy_pct": pct(d["m5"])},
|
|
"m15": {**d["m15"], "accuracy_pct": pct(d["m15"])},
|
|
"m1h": {**d["m1h"], "accuracy_pct": pct(d["m1h"])},
|
|
}
|
|
for s, d in by_signal.items()
|
|
},
|
|
"total_directional_signals": len(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)
|