"""智策股票终端 — FastAPI 后端入口。

- /api/*  : 数据接口（基于 AkShare，带缓存与降级）
- /       : 托管前端原型（prototype 目录）
"""
import os
import json
import datetime as dt
from contextlib import asynccontextmanager

from fastapi import FastAPI, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from sqlalchemy import select, func, desc

from pydantic import BaseModel

import akshare_service as svc
import config
import scheduler
import backtest as bt
import ai
import signals as sig
import report as rpt
import portfolio as pf
import llm
import alerts as al
import notifier
from db import init_db, get_session
from models import (DailyQuote, IndexDaily, SectorDaily, FundFlowDaily,
                    SentimentDaily, DragonTiger, Security, JobRun, StockMetric, Trade,
                    AlertRule, AlertEvent)


@asynccontextmanager
async def lifespan(app: FastAPI):
    try:
        init_db()
        scheduler.start_scheduler()
        print("[startup] db + scheduler ready")
    except Exception as e:
        print("[startup] WARN:", repr(e)[:160])
    yield


app = FastAPI(title="智策股票终端 API", version="0.2.0", lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 自选股本地存储
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
WATCH_FILE = os.path.join(BASE_DIR, "watchlist.json")
DEFAULT_WATCH = ["600519", "300750", "002594", "688981", "300059", "601012"]


def load_watch():
    if os.path.exists(WATCH_FILE):
        try:
            with open(WATCH_FILE, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            pass
    return DEFAULT_WATCH


def save_watch(symbols):
    with open(WATCH_FILE, "w", encoding="utf-8") as f:
        json.dump(symbols, f, ensure_ascii=False)


# ============ API ============
@app.get("/api/health")
def health():
    return {"ok": True, "akshare": svc.AK_OK}


@app.get("/api/indices")
def indices():
    return svc.get_indices()


@app.get("/api/kline")
def kline(symbol: str = Query("600519"), days: int = Query(120, ge=20, le=500)):
    return svc.get_kline(symbol, days)


@app.get("/api/sentiment")
def sentiment():
    return svc.get_sentiment()


@app.get("/api/treemap")
def treemap(mode: str = Query("sector")):
    return svc.get_treemap(mode)


@app.get("/api/fundflow")
def fundflow():
    return svc.get_fund_flow()


@app.get("/api/hot/stocks")
def hot_stocks():
    return svc.get_hot_stocks()


@app.get("/api/hot/sectors")
def hot_sectors():
    return svc.get_industry_boards()


@app.get("/api/dragon")
def dragon():
    return svc.get_dragon_tiger()


@app.get("/api/watchlist")
def watchlist():
    return svc.get_watchlist(load_watch())


@app.post("/api/watchlist/{code}")
def watch_add(code: str):
    w = load_watch()
    if code not in w:
        w.append(code)
        save_watch(w)
    return {"ok": True, "list": w}


@app.delete("/api/watchlist/{code}")
def watch_del(code: str):
    w = [c for c in load_watch() if c != code]
    save_watch(w)
    return {"ok": True, "list": w}


# ============ 数据中台 ============
@app.get("/api/admin/status")
def admin_status():
    counts, last_dates = {}, {}
    with get_session() as s:
        for label, model in [("securities", Security), ("quotes_daily", DailyQuote),
                             ("index_daily", IndexDaily), ("sector_daily", SectorDaily),
                             ("fund_flow_daily", FundFlowDaily), ("sentiment_daily", SentimentDaily),
                             ("dragon_tiger", DragonTiger)]:
            counts[label] = s.execute(select(func.count()).select_from(model)).scalar() or 0
            if hasattr(model, "date"):
                d = s.execute(select(func.max(model.date))).scalar()
                last_dates[label] = d.isoformat() if d else None
        jobs = s.execute(select(JobRun).order_by(desc(JobRun.id)).limit(8)).scalars().all()
        job_list = [{"id": j.id, "job": j.job, "status": j.status,
                     "started": j.started_at.strftime("%m-%d %H:%M:%S") if j.started_at else "",
                     "finished": j.finished_at.strftime("%H:%M:%S") if j.finished_at else "",
                     "message": j.message[:200]} for j in jobs]
    return {"counts": counts, "last_dates": last_dates, "jobs": job_list,
            "running": scheduler.is_running(), "universe": config.DEFAULT_UNIVERSE,
            "schedule": f"周一至周五 {config.INGEST_HOUR:02d}:{config.INGEST_MINUTE:02d}"}


@app.post("/api/admin/ingest")
def admin_ingest():
    if scheduler.is_running():
        return {"started": False, "msg": "已有入库任务在执行"}
    return scheduler.trigger_async()


@app.post("/api/admin/ingest_all")
def admin_ingest_all():
    return scheduler.trigger_all_async()


@app.get("/api/db/kline")
def db_kline(symbol: str = Query("600519"), days: int = Query(250, ge=20, le=1000)):
    with get_session() as s:
        rows = s.execute(
            select(DailyQuote).where(DailyQuote.code == symbol)
            .order_by(DailyQuote.date.desc()).limit(days)
        ).scalars().all()
    rows = list(reversed(rows))
    if not rows:
        return {"source": "db", "empty": True, "symbol": symbol, "dates": [], "ohlc": [], "vols": []}
    return {"source": "db", "symbol": symbol,
            "dates": [r.date.strftime("%m/%d") for r in rows],
            "ohlc": [[r.open, r.close, r.low, r.high] for r in rows],
            "vols": [r.volume for r in rows]}


@app.get("/api/db/sentiment_history")
def db_sentiment_history(days: int = Query(60, ge=5, le=365)):
    with get_session() as s:
        rows = s.execute(select(SentimentDaily).order_by(SentimentDaily.date.desc()).limit(days)).scalars().all()
    rows = list(reversed(rows))
    return {"dates": [r.date.isoformat() for r in rows],
            "up": [r.up for r in rows], "down": [r.down for r in rows],
            "limit_up": [r.limit_up for r in rows]}


@app.get("/api/review/daily")
def review_daily(date: str = Query(None)):
    with get_session() as s:
        if date:
            d = dt.date.fromisoformat(date)
        else:
            d = s.execute(select(func.max(SectorDaily.date))).scalar()
        if not d:
            return {"ok": False, "msg": "暂无入库数据，请先在数据中台执行入库"}
        sectors = s.execute(select(SectorDaily).where(SectorDaily.date == d).order_by(SectorDaily.pct.desc())).scalars().all()
        flows = s.execute(select(FundFlowDaily).where(FundFlowDaily.date == d).order_by(FundFlowDaily.net.desc())).scalars().all()
        senti = s.execute(select(SentimentDaily).where(SentimentDaily.date == d)).scalar_one_or_none()
        lhb = s.execute(select(DragonTiger).where(DragonTiger.date == d).order_by(DragonTiger.net.desc()).limit(10)).scalars().all()

    top_sec = [{"name": x.name, "pct": x.pct} for x in sectors[:8]]
    bot_sec = [{"name": x.name, "pct": x.pct} for x in sectors[-5:]]
    inflow = [{"name": x.name, "net": x.net} for x in flows[:8]]
    outflow = [{"name": x.name, "net": x.net} for x in flows[-5:][::-1]]
    senti_d = ({"up": senti.up, "down": senti.down, "limit_up": senti.limit_up,
                "limit_down": senti.limit_down} if senti else None)

    summary = _gen_review_text(d, senti_d, top_sec, inflow)
    return {"ok": True, "date": d.isoformat(), "sentiment": senti_d,
            "top_sectors": top_sec, "weak_sectors": bot_sec,
            "inflow": inflow, "outflow": outflow,
            "dragon": [{"name": x.name, "code": x.code, "net": x.net, "pct": x.pct} for x in lhb],
            "summary": summary}


def _gen_review_text(d, senti, top_sec, inflow):
    parts = [f"【{d.isoformat()} 复盘】"]
    if senti:
        tone = "情绪偏暖" if senti["up"] > senti["down"] else "情绪偏弱"
        parts.append(f"全市场上涨 {senti['up']} 家、下跌 {senti['down']} 家，涨停 {senti['limit_up']} 家、跌停 {senti['limit_down']} 家，{tone}。")
    if top_sec:
        names = "、".join(x["name"] for x in top_sec[:3])
        parts.append(f"领涨板块：{names}。")
    if inflow:
        names = "、".join(x["name"] for x in inflow[:3])
        parts.append(f"主力净流入居前：{names}。")
    parts.append("注：以上为基于入库数据的自动统计，AI 智能点评将在 AI 分析模块接入大模型后生成。")
    return " ".join(parts)


@app.get("/api/backtest")
def backtest_api(symbol: str = Query("600519"), fast: int = Query(5, ge=2, le=60),
                 slow: int = Query(20, ge=5, le=250)):
    if fast >= slow:
        return {"ok": False, "msg": "快线周期需小于慢线周期"}
    return bt.run_backtest(symbol, fast, slow)


# ============ 全市场选股 ============
STRATEGIES = {
    "surge": "最近暴涨（5日涨幅≥20%）",
    "plunge": "最近暴跌（5日跌幅≥15%）",
    "dip": "超跌抄底（60日分位≤20%且当日企稳）",
    "breakout": "突破走强（逼近60日新高）",
    "ma_bull": "均线多头（MA5>10>20）",
    "volume": "放量上攻（量比≥2且上涨）",
    "macd_gold": "MACD金叉",
    "strong": "强势连涨（≥3日连阳）",
}


@app.get("/api/screen/strategies")
def screen_strategies():
    return {"list": [{"id": k, "name": v} for k, v in STRATEGIES.items()]}


@app.get("/api/screen")
def screen(strategy: str = Query("surge"), limit: int = Query(60, ge=10, le=300),
           min_amount: float = Query(0.0)):
    M = StockMetric
    q = select(M)
    order = M.ret5.desc()
    if strategy == "surge":
        q = q.where(M.ret5 >= 20)
    elif strategy == "plunge":
        q = q.where(M.ret5 <= -15); order = M.ret5.asc()
    elif strategy == "dip":
        q = q.where(M.pos60 <= 0.2, M.pct > 0); order = M.pos60.asc()
    elif strategy == "breakout":
        q = q.where(M.pos60 >= 0.95, M.pct > 0); order = M.ret20.desc()
    elif strategy == "ma_bull":
        q = q.where(M.ma_bull.is_(True)); order = M.ret20.desc()
    elif strategy == "volume":
        q = q.where(M.vol_ratio >= 2, M.pct > 0); order = M.vol_ratio.desc()
    elif strategy == "macd_gold":
        q = q.where(M.macd_gold.is_(True)); order = M.ret5.desc()
    elif strategy == "strong":
        q = q.where(M.up_streak >= 3); order = M.up_streak.desc()
    if min_amount > 0:
        q = q.where(M.amount >= min_amount)
    q = q.order_by(order).limit(limit)
    with get_session() as s:
        rows = s.execute(q).scalars().all()
        total = s.execute(select(func.count()).select_from(M)).scalar() or 0
    return {"strategy": strategy, "name": STRATEGIES.get(strategy, strategy), "pool_size": total,
            "count": len(rows), "list": [{
                "code": r.code, "name": r.name, "close": r.close, "pct": r.pct,
                "ret5": r.ret5, "ret20": r.ret20, "vol_ratio": r.vol_ratio,
                "rsi14": r.rsi14, "pos60": round(r.pos60 * 100, 1), "amount": r.amount,
                "up_streak": r.up_streak} for r in rows]}


@app.get("/api/securities/search")
def securities_search(q: str = Query("", min_length=0), limit: int = Query(15, le=50)):
    with get_session() as s:
        stmt = select(Security)
        if q:
            stmt = stmt.where((Security.code.like(f"{q}%")) | (Security.name.like(f"%{q}%")))
        rows = s.execute(stmt.limit(limit)).scalars().all()
    return {"list": [{"code": r.code, "name": r.name} for r in rows]}


# ============ 个股复盘（K线 + 买卖点 + 回放） ============
def _ma_list(close, n):
    out = [None] * len(close)
    for i in range(len(close)):
        if i >= n - 1:
            out[i] = round(sum(close[i - n + 1:i + 1]) / n, 3)
    return out


@app.get("/api/review/stock")
def review_stock(symbol: str = Query("600519"), days: int = Query(250, ge=40, le=1000),
                 fast: int = Query(5), slow: int = Query(20)):
    with get_session() as s:
        rows = s.execute(
            select(DailyQuote).where(DailyQuote.code == symbol)
            .order_by(DailyQuote.date.desc()).limit(days)
        ).scalars().all()
        sec = s.get(Security, symbol)
    rows = list(reversed(rows))
    if not rows:
        return {"ok": False, "msg": "该股票库内无日线，请先在数据中台入库该股或执行全市场回填", "symbol": symbol}
    dates = [r.date.strftime("%y/%m/%d") for r in rows]
    ohlc = [[r.open, r.close, r.low, r.high] for r in rows]
    vols = [r.volume for r in rows]
    close = [r.close for r in rows]
    maf, mas = _ma_list(close, fast), _ma_list(close, slow)

    signals = []
    for i in range(1, len(close)):
        if maf[i] is None or mas[i] is None or maf[i - 1] is None or mas[i - 1] is None:
            continue
        if maf[i - 1] <= mas[i - 1] and maf[i] > mas[i]:
            signals.append({"idx": i, "date": dates[i], "price": close[i], "type": "buy"})
        elif maf[i - 1] >= mas[i - 1] and maf[i] < mas[i]:
            signals.append({"idx": i, "date": dates[i], "price": close[i], "type": "sell"})

    # 区间统计
    hi = max(r.high for r in rows); lo = min(r.low for r in rows)
    period_ret = round((close[-1] / close[0] - 1) * 100, 2)
    return {"ok": True, "symbol": symbol, "name": sec.name if sec else symbol,
            "dates": dates, "ohlc": ohlc, "vols": vols,
            "ma_fast": maf, "ma_slow": mas, "fast": fast, "slow": slow,
            "signals": signals,
            "stats": {"period_return": period_ret, "high": hi, "low": lo,
                      "start": dates[0], "end": dates[-1], "bars": len(rows)}}


# ============ AI 分析 ============
@app.get("/api/ai/status")
def ai_status():
    return {"enabled": llm.enabled(), "model": config.LLM_MODEL if llm.enabled() else None}


@app.get("/api/ai/review_daily")
def ai_review_daily(date: str = Query(None)):
    return ai.review_daily_comment(date)


@app.get("/api/ai/diagnose")
def ai_diagnose(symbol: str = Query("600519")):
    return ai.diagnose(symbol)


@app.get("/api/ai/today")
def ai_today():
    return ai.today_strategy()


# ============ 可回溯：信号历史胜率 + 实测准确率 ============
@app.get("/api/ai/signal_stats")
def ai_signal_stats(horizon: int = Query(5, ge=1, le=20)):
    return {"horizon": horizon, "stats": sig.get_stats(horizon)}


@app.post("/api/ai/signal_stats/compute")
def ai_signal_stats_compute(sample: int = Query(500, ge=50, le=4000), horizon: int = Query(5, ge=1, le=20)):
    return scheduler.trigger_signal_stats_async(sample, horizon)


@app.get("/api/ai/accuracy")
def ai_accuracy():
    return sig.accuracy()


@app.post("/api/ai/accuracy/verify")
def ai_accuracy_verify():
    return sig.verify_predictions()


# ============ AI 自动复盘日报 ============
@app.get("/api/report/daily")
def report_daily(date: str = Query(None)):
    return rpt.get_by_date(date) if date else rpt.latest()


@app.get("/api/report/history")
def report_history(limit: int = Query(30, ge=1, le=120)):
    return rpt.history(limit)


@app.post("/api/report/generate")
def report_generate(date: str = Query(None), push: bool = Query(False)):
    return rpt.generate(date, push=push)


# ============ 交易日志 & 组合 ============
class TradeIn(BaseModel):
    code: str
    name: str = ""
    side: str = "buy"
    price: float
    qty: int
    fee: float = 0.0
    date: str = ""
    reason: str = ""
    emotion: str = ""


@app.get("/api/trades")
def list_trades():
    with get_session() as s:
        rows = s.execute(select(Trade).order_by(Trade.date.desc(), Trade.id.desc())).scalars().all()
        names = {}
    return {"list": [{"id": t.id, "date": t.date.isoformat(), "code": t.code, "name": t.name,
                      "side": t.side, "price": t.price, "qty": t.qty, "fee": t.fee,
                      "reason": t.reason, "emotion": t.emotion} for t in rows]}


@app.post("/api/trades")
def add_trade(t: TradeIn):
    d = dt.date.fromisoformat(t.date) if t.date else dt.date.today()
    name = t.name
    if not name:
        with get_session() as s:
            sec = s.get(Security, t.code)
            name = sec.name if sec else t.code
    with get_session() as s:
        row = Trade(date=d, code=t.code, name=name, side=t.side, price=t.price,
                    qty=t.qty, fee=t.fee, reason=t.reason, emotion=t.emotion)
        s.add(row); s.commit()
        return {"ok": True, "id": row.id}


@app.delete("/api/trades/{tid}")
def del_trade(tid: int):
    with get_session() as s:
        row = s.get(Trade, tid)
        if row:
            s.delete(row); s.commit()
    return {"ok": True}


@app.get("/api/portfolio")
def get_portfolio():
    return pf.compute()


@app.get("/api/portfolio/equity")
def portfolio_equity():
    return pf.equity_curve()


# ============ 推送通知 ============
@app.get("/api/notify/status")
def notify_status():
    return {"channels": notifier.channels_status(), "enabled": notifier.any_enabled()}


@app.post("/api/notify/test")
def notify_test():
    if not notifier.any_enabled():
        return {"ok": False, "msg": "未配置任何推送渠道，请在 backend/.env 配置后重启"}
    res = notifier.notify("【智策】推送测试", "这是一条来自智策股票终端的测试通知，收到即表示推送通道正常。")
    return {"ok": True, "result": res}


# ============ 智能预警 ============
class AlertIn(BaseModel):
    code: str
    kind: str = "price_above"
    threshold: float
    note: str = ""


@app.get("/api/alerts")
def list_alerts():
    with get_session() as s:
        rows = s.execute(select(AlertRule).order_by(AlertRule.id.desc())).scalars().all()
    return {"list": [{"id": r.id, "code": r.code, "name": r.name, "kind": r.kind,
                      "threshold": r.threshold, "status": r.status, "note": r.note,
                      "last_value": r.last_value,
                      "triggered_at": r.triggered_at.strftime("%m-%d %H:%M") if r.triggered_at else ""}
                     for r in rows]}


@app.post("/api/alerts")
def add_alert(a: AlertIn):
    with get_session() as s:
        sec = s.get(Security, a.code)
        name = sec.name if sec else a.code
        row = AlertRule(code=a.code, name=name, kind=a.kind, threshold=a.threshold, note=a.note)
        s.add(row); s.commit()
        return {"ok": True, "id": row.id}


@app.delete("/api/alerts/{aid}")
def del_alert(aid: int):
    with get_session() as s:
        row = s.get(AlertRule, aid)
        if row:
            s.delete(row); s.commit()
    return {"ok": True}


@app.post("/api/alerts/{aid}/reactivate")
def reactivate_alert(aid: int):
    with get_session() as s:
        row = s.get(AlertRule, aid)
        if row:
            row.status = "active"; row.triggered_at = None; s.commit()
    return {"ok": True}


@app.post("/api/alerts/check")
def manual_check():
    return al.check_alerts()


@app.get("/api/alerts/events")
def alert_events(unread_only: bool = Query(False), limit: int = Query(30, le=100)):
    with get_session() as s:
        stmt = select(AlertEvent).order_by(AlertEvent.id.desc())
        if unread_only:
            stmt = stmt.where(AlertEvent.read.is_(False))
        rows = s.execute(stmt.limit(limit)).scalars().all()
        unread = s.execute(select(func.count()).select_from(AlertEvent).where(AlertEvent.read.is_(False))).scalar() or 0
    return {"unread": unread, "list": [{"id": e.id, "code": e.code, "name": e.name,
            "message": e.message, "time": e.created_at.strftime("%m-%d %H:%M:%S") if e.created_at else ""}
            for e in rows]}


@app.post("/api/alerts/events/read")
def mark_events_read():
    with get_session() as s:
        for e in s.execute(select(AlertEvent).where(AlertEvent.read.is_(False))).scalars():
            e.read = True
        s.commit()
    return {"ok": True}


# ============ 资讯中心 ============
@app.get("/api/news")
def news(limit: int = Query(40, le=100)):
    return svc.get_news(limit)


@app.get("/api/news/stock")
def news_stock(code: str = Query(...)):
    return svc.get_stock_news(code)


@app.get("/api/news/watch")
def news_watch():
    codes = load_watch()[:6]
    out = []
    for c in codes:
        r = svc.get_stock_news(c, limit=4)
        for x in r["list"]:
            x["code"] = c
            out.append(x)
    out.sort(key=lambda x: x["time"], reverse=True)
    return {"list": out[:40]}


class NewsAI(BaseModel):
    title: str
    content: str = ""


@app.post("/api/news/ai")
def news_ai(n: NewsAI):
    text_in = (n.title + "。" + n.content).strip()
    senti, kw = svc.judge_sentiment(text_in)
    if llm.enabled():
        try:
            prompt = ("请分析下面这条财经资讯：\n"
                      "1) 一句话摘要；2) 利好/利空/中性判断及理由；3) 可能受影响的板块或个股方向。120字内。\n\n"
                      + text_in[:1200])
            text = llm.ask(prompt, temperature=0.3, max_tokens=400)
            return {"ok": True, "source": "llm", "sentiment": senti, "text": text}
        except Exception:
            pass
    return {"ok": True, "source": "rule", "sentiment": senti,
            "text": f"判断：{senti}（关键词：{'、'.join(kw) or '无'}）。摘要：{text_in[:80]}…\n（配置大模型后可获得更深入的关联分析）"}


# ============ 静态前端 ============
FRONTEND_DIR = os.path.join(os.path.dirname(BASE_DIR), "prototype")
if os.path.isdir(FRONTEND_DIR):
    app.mount("/", StaticFiles(directory=FRONTEND_DIR, html=True), name="frontend")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=False)