"""Data collector — fetch OHLCV from akshare and upsert into daily_bars."""
import threading
import time
from datetime import date
from app.database import SessionLocal
from app.models import DailyBar, Contract
from app.engine.lock_strategy import compute_amp_5d

_progress = {"running": False, "label": "", "done": 0, "total": 0, "result": 0, "finished": False, "error": "", "started_at": 0}
_lock = threading.Lock()

STALE_SECONDS = 120


def get_progress() -> dict:
    with _lock:
        return dict(_progress)


def _start_bg(target, label):
    with _lock:
        now = time.time()
        if _progress["running"] and (now - _progress["started_at"]) < STALE_SECONDS:
            return False
        _progress.update(running=True, finished=False, label=label, done=0, total=0, error="", started_at=now)

    def _run():
        try:
            target()
        except Exception as e:
            with _lock:
                _progress["error"] = str(e)
        finally:
            with _lock:
                _progress["running"] = False
                _progress["finished"] = True
                _progress["done"] = _progress["total"]

    threading.Thread(target=_run, daemon=True).start()
    return True


def sync_contract_bars_bg(contract_code: str) -> bool:
    """Sync bars for one contract in background. Returns True if started."""
    code = contract_code.upper()

    def _run():
        db = SessionLocal()
        try:
            latest = (
                db.query(DailyBar.date)
                .filter(DailyBar.contract == code)
                .order_by(DailyBar.date.desc())
                .first()
            )
            start_date = latest[0].isoformat() if latest else None
            bars = fetch_contract_bars(code, start_date)

            with _lock:
                _progress["total"] = len(bars)

            total_bars = len(bars)
            inserted = 0
            min_date = None
            for i, bar in enumerate(bars):
                existing = (
                    db.query(DailyBar)
                    .filter(DailyBar.contract == code, DailyBar.date == bar["date"])
                    .first()
                )
                if not existing:
                    db.add(DailyBar(
                        contract=code, date=bar["date"],
                        open=bar["open"], close=bar["close"],
                        high=bar["high"], low=bar["low"],
                    ))
                    inserted += 1
                    if min_date is None or bar["date"] < min_date:
                        min_date = bar["date"]
                with _lock:
                    _progress["done"] = i + 1
                    _progress["label"] = f"同步行情 {code} · {i + 1}/{total_bars} 条"

            if inserted > 0:
                db.flush()
                _recompute_amp(db, code, from_date=min_date)
            db.commit()
            with _lock:
                _progress["result"] = inserted
        finally:
            db.close()

    return _start_bg(_run, f"同步行情 {code}")


def sync_positions_bg(contract_code: str) -> bool:
    """Sync position rankings for one contract in background. Returns True if started."""
    from app.models import PositionRanking
    code = contract_code.upper()

    def _run():
        db = SessionLocal()
        try:
            existing_dates = {
                r[0] for r in
                db.query(PositionRanking.date)
                .filter(PositionRanking.contract_code == code)
                .distinct().all()
            }
            bar_dates = [
                r[0] for r in
                db.query(DailyBar.date)
                .filter(DailyBar.contract == code)
                .order_by(DailyBar.date).all()
            ]
            missing = [d for d in bar_dates if d not in existing_dates]
            total_missing = len(missing)

            with _lock:
                _progress["total"] = total_missing
                if total_missing == 0:
                    _progress["label"] = f"同步持仓 {code} · 已是最新"

            inserted = 0
            for i, d in enumerate(missing):
                rankings = fetch_position_rankings(code, d.strftime("%Y%m%d"))
                for r in rankings:
                    db.add(PositionRanking(
                        contract_code=code, institution=r["institution"],
                        data_type=r["data_type"], date=d,
                        rank=r["rank"], value=r["value"], change=r["change"],
                    ))
                    inserted += 1
                db.flush()
                with _lock:
                    _progress["done"] = i + 1
                    _progress["label"] = f"同步持仓 {code} · {i + 1}/{total_missing} 天"
            db.commit()
            with _lock:
                _progress["result"] = inserted
        finally:
            db.close()

    return _start_bg(_run, f"同步持仓 {code}")


def fetch_contract_bars(contract_code: str, start_date: str | None = None) -> list[dict]:
    """Fetch daily OHLCV for a single contract from akshare.

    Returns list of {date, open, close, high, low} dicts.
    Only returns bars on or after start_date when provided.
    """
    import akshare as ak

    try:
        df = ak.futures_zh_daily_sina(symbol=contract_code.upper())
    except Exception as e:
        print(f"[collector] akshare error for {contract_code}: {e}")
        return []

    if df is None or df.empty:
        print(f"[collector] No data returned for {contract_code}")
        return []

    filter_date = date.fromisoformat(start_date) if start_date else None

    bars = []
    for _, row in df.iterrows():
        try:
            val = row["date"]
            if hasattr(val, "strftime"):
                d = val.date() if hasattr(val, "date") else val
            else:
                d = date.fromisoformat(str(val)[:10])

            if filter_date and d < filter_date:
                continue

            bars.append({
                "date": d,
                "open": float(row["open"]),
                "close": float(row["close"]),
                "high": float(row["high"]),
                "low": float(row["low"]),
            })
        except (KeyError, ValueError, TypeError) as e:
            print(f"[collector] skip row: {e}")
            continue

    return bars


def sync_active_contracts() -> dict:
    """Sync all active contracts. Returns {contract_code: new_bars_count}."""
    db = SessionLocal()
    results = {}

    try:
        active_contracts = (
            db.query(Contract).filter(Contract.is_active == True).all()
        )

        for c in active_contracts:
            count = _sync_one(db, c.code)
            results[c.code] = count

        db.commit()
    finally:
        db.close()

    return results


def sync_one_contract(contract_code: str) -> int:
    """Sync a single contract. Returns number of new bars inserted."""
    db = SessionLocal()
    try:
        count = _sync_one(db, contract_code.upper())
        db.commit()
        return count
    finally:
        db.close()


def _sync_one(db, contract_code: str) -> int:
    """Internal: sync one contract using existing session."""
    latest = (
        db.query(DailyBar.date)
        .filter(DailyBar.contract == contract_code)
        .order_by(DailyBar.date.desc())
        .first()
    )
    start_date = latest[0].isoformat() if latest else None

    bars = fetch_contract_bars(contract_code, start_date)
    if not bars:
        return 0

    inserted = 0
    min_date = None
    for bar in bars:
        existing = (
            db.query(DailyBar)
            .filter(
                DailyBar.contract == contract_code,
                DailyBar.date == bar["date"],
            )
            .first()
        )
        if not existing:
            db.add(DailyBar(
                contract=contract_code,
                date=bar["date"],
                open=bar["open"],
                close=bar["close"],
                high=bar["high"],
                low=bar["low"],
            ))
            inserted += 1
            if min_date is None or bar["date"] < min_date:
                min_date = bar["date"]

    if inserted > 0:
        db.flush()
        _recompute_amp(db, contract_code, from_date=min_date)

    return inserted


def fetch_position_rankings(contract_code: str, trade_date: str) -> list[dict]:
    """Fetch top-20 position rankings for a contract on a given date from akshare.

    Calls the API 3 times (volume, long, short) and returns a unified list of dicts:
    {data_type, rank, institution, value, change}
    """
    import akshare as ak

    results = []
    for sym, dtype in [("成交量", "volume"), ("多单持仓", "long"), ("空单持仓", "short")]:
        try:
            df = ak.futures_hold_pos_sina(symbol=sym, contract=contract_code.upper(), date=trade_date)
        except Exception as e:
            print(f"[collector] position akshare error for {contract_code} {dtype}: {e}")
            continue

        if df is None or df.empty:
            continue

        for _, row in df.iterrows():
            try:
                results.append({
                    "data_type": dtype,
                    "rank": int(row["名次"]),
                    "institution": str(row["会员简称"]),
                    "value": int(row.iloc[2]),
                    "change": int(row["比上交易增减"]),
                })
            except (KeyError, ValueError, TypeError) as e:
                print(f"[collector] position skip row: {e}")
                continue

    return results


def sync_position_rankings() -> dict:
    """Sync position rankings for all active contracts. Returns {contract_code: new_rows}."""
    db = SessionLocal()
    results = {}

    try:
        active_contracts = (
            db.query(Contract).filter(Contract.is_active == True).all()
        )

        for c in active_contracts:
            count = _sync_positions_for_contract(db, c.code)
            results[c.code] = count

        db.commit()
    finally:
        db.close()

    return results


def _sync_positions_for_contract(db, contract_code: str) -> int:
    """Sync position rankings for all dates that have bars but no position data."""
    from app.models import PositionRanking

    code = contract_code.upper()

    existing_dates = {
        r[0] for r in
        db.query(PositionRanking.date)
        .filter(PositionRanking.contract_code == code)
        .distinct()
        .all()
    }

    bar_dates = [
        r[0] for r in
        db.query(DailyBar.date)
        .filter(DailyBar.contract == code)
        .order_by(DailyBar.date)
        .all()
    ]

    inserted = 0
    for d in bar_dates:
        if d in existing_dates:
            continue
        date_str = d.strftime("%Y%m%d")
        rankings = fetch_position_rankings(code, date_str)
        for r in rankings:
            db.add(PositionRanking(
                contract_code=code,
                institution=r["institution"],
                data_type=r["data_type"],
                date=d,
                rank=r["rank"],
                value=r["value"],
                change=r["change"],
            ))
            inserted += 1
        db.flush()

    return inserted


def _recompute_amp(db, contract_code: str, from_date: date | None = None):
    """Recompute amp_5d for bars of a contract from from_date onwards."""
    bars = (
        db.query(DailyBar)
        .filter(DailyBar.contract == contract_code)
        .order_by(DailyBar.date)
        .all()
    )
    start_idx = 0
    if from_date:
        for i, bar in enumerate(bars):
            if bar.date >= from_date:
                start_idx = i
                break
    for i, bar in enumerate(bars):
        if i >= 5 and i >= start_idx:
            bar.amp_5d = compute_amp_5d([b.diff for b in bars[i - 5 : i]])