# Copyright (c) 2025-2026 马建军. All rights reserved.
# 专有软件 — 未经授权禁止复制、传播、转售。
# 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。
# 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md

"""账户冷静期 / 日冻结（自 crypto_monitor 复制并简化为单账户期货版）。"""
from __future__ import annotations

import os
import time
from datetime import datetime
from typing import Any, Callable, Optional, TypeVar
from zoneinfo import ZoneInfo

from db_conn import OperationalError, is_missing_relation_error, rollback_if_postgres

T = TypeVar("T")

STATUS_NORMAL = "normal"
STATUS_FREEZE_1H = "freeze_1h"
STATUS_FREEZE_4H = "freeze_4h"
STATUS_DAILY = "freeze_daily"
STATUS_FREEZE_POSITION = "freeze_position"

STATUS_LABELS = {
    STATUS_NORMAL: "正常",
    STATUS_FREEZE_1H: "1h冻结",
    STATUS_FREEZE_4H: "4h冻结",
    STATUS_DAILY: "日冻结",
    STATUS_FREEZE_POSITION: "仓位上限冻结",
}

MOOD_ISSUE_OPTIONS = (
    "怕踏空", "报复开仓", "盈利飘了", "拿不住单", "扛单", "重仓违规",
)

CLOSE_SOURCE_USER = "user_instance"
CLOSE_SOURCE_TREND_STOP = "user_trend_stop"


def _app_tz():
    name = (os.getenv("APP_TIMEZONE") or "Asia/Shanghai").strip()
    try:
        return ZoneInfo(name)
    except Exception:
        return ZoneInfo("Asia/Shanghai")


def risk_control_enabled() -> bool:
    raw = (os.getenv("RISK_CONTROL_ENABLED") or "true").strip().lower()
    return raw in ("1", "true", "yes", "on")


def cooling_hours_manual() -> float:
    """期货版不使用应用层冷静期（交易所自有规则），恒为 0。"""
    return 0.0


def cooling_hours_manual_journal() -> float:
    return 0.0


def manual_close_daily_limit() -> int:
    try:
        return max(1, int(os.getenv("RISK_MANUAL_CLOSE_DAILY_LIMIT", "2")))
    except (TypeError, ValueError):
        return 2


def max_active_positions() -> int:
    try:
        return max(1, int(os.getenv("MAX_ACTIVE_POSITIONS", "1")))
    except (TypeError, ValueError):
        return 1


def daily_position_limit() -> int:
    """当日最多开仓次数（含已平）。"""
    try:
        return max(1, int(os.getenv("RISK_DAILY_POSITION_LIMIT", "5")))
    except (TypeError, ValueError):
        return 5


def daily_trading_risk_pct_limit() -> float:
    """当日累计止损风险占权益上限（%）。"""
    try:
        return max(0.1, float(os.getenv("RISK_DAILY_TRADING_RISK_PCT", "2")))
    except (TypeError, ValueError):
        return 2.0


def trading_day_reset_hour() -> int:
    try:
        return max(0, min(23, int(os.getenv("TRADING_DAY_RESET_HOUR", "8"))))
    except (TypeError, ValueError):
        return 8


_SCHEMA_READY = False

ACCOUNT_RISK_STATE_SQL = """
CREATE TABLE IF NOT EXISTS account_risk_state (
    id INTEGER PRIMARY KEY,
    trading_day TEXT,
    manual_close_count INTEGER DEFAULT 0,
    cooloff_until_ms INTEGER,
    cooloff_hours INTEGER,
    daily_frozen INTEGER DEFAULT 0,
    last_close_at_ms INTEGER,
    updated_at TEXT
)
"""


def _account_risk_table_exists(conn) -> bool:
    try:
        conn.execute("SELECT 1 FROM account_risk_state WHERE id=1")
        return True
    except Exception as exc:
        if is_missing_relation_error(exc):
            rollback_if_postgres(conn)
            return False
        raise


def _db_retry(action: Callable[[], T], *, retries: int = 8, base_delay: float = 0.03) -> T:
    last: OperationalError | None = None
    for i in range(retries):
        try:
            return action()
        except OperationalError as exc:
            msg = str(exc).lower()
            if "locked" not in msg and "serialize" not in msg and "deadlock" not in msg:
                raise
            last = exc
            time.sleep(base_delay * (2 ** i))
    if last is not None:
        raise last
    raise RuntimeError("db retry failed")


def ensure_account_risk_schema(conn) -> None:
    global _SCHEMA_READY
    if _SCHEMA_READY and _account_risk_table_exists(conn):
        return
    _SCHEMA_READY = False
    conn.execute(ACCOUNT_RISK_STATE_SQL)
    conn.commit()
    if not conn.execute("SELECT 1 FROM account_risk_state WHERE id=1").fetchone():
        conn.execute(
            "INSERT INTO account_risk_state (id, trading_day, manual_close_count, daily_frozen) "
            "VALUES (1, '', 0, 0)"
        )
    conn.commit()
    _SCHEMA_READY = True


def _row_get(row, key, default=None):
    if row is None:
        return default
    try:
        return row[key]
    except (KeyError, IndexError, TypeError):
        return default


def _now_ms(now: Optional[datetime] = None) -> int:
    dt = now or datetime.now(_app_tz())
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=_app_tz())
    return int(dt.timestamp() * 1000)


def trading_day_label(now: Optional[datetime] = None) -> str:
    dt = now or datetime.now(_app_tz())
    if dt.hour < trading_day_reset_hour():
        from datetime import timedelta
        dt = dt - timedelta(days=1)
    return dt.date().isoformat()


def trading_day_start(now: Optional[datetime] = None) -> datetime:
    dt = now or datetime.now(_app_tz())
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=_app_tz())
    reset_h = trading_day_reset_hour()
    start = dt.replace(hour=reset_h, minute=0, second=0, microsecond=0)
    if dt.hour < reset_h:
        from datetime import timedelta
        start = start - timedelta(days=1)
    return start


def _parse_open_time_ms(open_time: str) -> Optional[int]:
    s = (open_time or "").strip().replace("T", " ")[:19]
    if not s:
        return None
    try:
        dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=_app_tz())
        return int(dt.timestamp() * 1000)
    except ValueError:
        try:
            dt = datetime.strptime(s[:10], "%Y-%m-%d").replace(tzinfo=_app_tz())
            return int(dt.timestamp() * 1000)
        except ValueError:
            return None


def _opened_in_trading_day(open_time: str, now: Optional[datetime] = None) -> bool:
    oms = _parse_open_time_ms(open_time)
    if oms is None:
        return False
    return oms >= int(trading_day_start(now).timestamp() * 1000)


def count_daily_opens(conn, now: Optional[datetime] = None) -> int:
    rows = conn.execute(
        "SELECT open_time FROM trade_order_monitors "
        "WHERE open_time IS NOT NULL AND trim(open_time) <> ''"
    ).fetchall()
    return sum(1 for r in rows if _opened_in_trading_day(r["open_time"], now))


def daily_trading_risk_used_pct(
    conn, equity: float, now: Optional[datetime] = None,
) -> Optional[float]:
    if equity <= 0:
        return None
    from contract_specs import calc_position_metrics

    total = 0.0
    rows = conn.execute(
        """SELECT symbol, direction, lots, entry_price, stop_loss, take_profit, open_time
           FROM trade_order_monitors
           WHERE open_time IS NOT NULL AND trim(open_time) <> ''"""
    ).fetchall()
    for r in rows:
        if not _opened_in_trading_day(r["open_time"], now):
            continue
        entry = float(r["entry_price"] or 0)
        if entry <= 0:
            continue
        sl = float(r["stop_loss"] if r["stop_loss"] is not None else entry)
        tp = float(r["take_profit"] if r["take_profit"] is not None else entry)
        lots = int(r["lots"] or 0)
        if lots <= 0:
            continue
        m = calc_position_metrics(
            r["direction"] or "long",
            entry,
            sl,
            tp,
            lots,
            entry,
            equity,
            r["symbol"] or "",
        )
        total += float(m.get("risk_amount") or 0)
    if total <= 0:
        return 0.0
    return round(total / equity * 100, 2)


def count_active_trade_monitors(conn) -> int:
    try:
        n = conn.execute(
            "SELECT COUNT(*) FROM trade_order_monitors WHERE status='active'"
        ).fetchone()[0]
        return int(n or 0)
    except Exception:
        return 0


def parse_mood_issues(raw: Any) -> list[str]:
    if raw is None:
        return []
    if isinstance(raw, (list, tuple)):
        parts = [str(x).strip() for x in raw if str(x).strip()]
    else:
        parts = [x.strip() for x in str(raw).split(",") if x.strip()]
    return [p for p in parts if p in MOOD_ISSUE_OPTIONS]


def on_user_initiated_close(conn, *, trading_day: str, now: Optional[datetime] = None) -> None:
    if not risk_control_enabled():
        return
    ensure_account_risk_schema(conn)
    row = conn.execute("SELECT * FROM account_risk_state WHERE id=1").fetchone()
    td = (trading_day or trading_day_label(now)).strip()
    stored = str(_row_get(row, "trading_day") or "")
    count = int(_row_get(row, "manual_close_count") or 0)
    if stored != td:
        count = 0
    count += 1
    close_ms = _now_ms(now)
    if count >= manual_close_daily_limit():
        conn.execute(
            """UPDATE account_risk_state SET trading_day=?, manual_close_count=?,
               daily_frozen=1, cooloff_until_ms=NULL, cooloff_hours=NULL,
               last_close_at_ms=?, updated_at=? WHERE id=1""",
            (td, count, close_ms, datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
        )
        return
    conn.execute(
        """UPDATE account_risk_state SET trading_day=?, manual_close_count=?,
           daily_frozen=0, cooloff_until_ms=NULL, cooloff_hours=NULL,
           last_close_at_ms=?, updated_at=? WHERE id=1""",
        (td, count, close_ms, datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
    )


def on_mood_journal_freeze(conn, *, trading_day: str) -> None:
    if not risk_control_enabled():
        return
    ensure_account_risk_schema(conn)
    td = (trading_day or trading_day_label()).strip()
    conn.execute(
        "UPDATE account_risk_state SET trading_day=?, daily_frozen=1, updated_at=? WHERE id=1",
        (td, datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
    )


def reduce_cooloff_after_journal(conn, *, trading_day: str, now: Optional[datetime] = None) -> None:
    """期货版无应用层冷静期，保留空实现兼容旧复盘钩子。"""
    del conn, trading_day, now
    return


def get_risk_status(
    conn,
    *,
    now: Optional[datetime] = None,
    active_count: Optional[int] = None,
    equity: Optional[float] = None,
) -> dict:
    def _load() -> dict:
        ensure_account_risk_schema(conn)
        try:
            row = conn.execute("SELECT * FROM account_risk_state WHERE id=1").fetchone()
        except Exception as exc:
            if is_missing_relation_error(exc):
                global _SCHEMA_READY
                _SCHEMA_READY = False
                rollback_if_postgres(conn)
                ensure_account_risk_schema(conn)
                row = conn.execute("SELECT * FROM account_risk_state WHERE id=1").fetchone()
            else:
                raise
        td = trading_day_label(now)
        stored = str(_row_get(row, "trading_day") or "")
        if stored != td:
            conn.execute(
                "UPDATE account_risk_state SET trading_day=?, manual_close_count=0, daily_frozen=0 WHERE id=1 AND trading_day<>?",
                (td, td),
            )
            conn.commit()
            row = conn.execute("SELECT * FROM account_risk_state WHERE id=1").fetchone()

        now_ms = _now_ms(now)
        daily = int(_row_get(row, "daily_frozen") or 0) == 1
        until = _row_get(row, "cooloff_until_ms")
        if until:
            conn.execute(
                "UPDATE account_risk_state SET cooloff_until_ms=NULL, cooloff_hours=NULL WHERE id=1"
            )
            conn.commit()
        active = count_active_trade_monitors(conn) if active_count is None else int(active_count)
        mx = max_active_positions()
        pos_limit = active >= mx
        daily_opens = count_daily_opens(conn, now)
        daily_pos_lim = daily_position_limit()
        daily_open_limit = daily_opens >= daily_pos_lim
        daily_risk_used: Optional[float] = None
        daily_risk_lim = daily_trading_risk_pct_limit()
        daily_risk_limit_hit = False
        if equity and float(equity) > 0:
            daily_risk_used = daily_trading_risk_used_pct(conn, float(equity), now)
            if daily_risk_used is not None and daily_risk_used >= daily_risk_lim:
                daily_risk_limit_hit = True

        base = {
            "active_count": active,
            "max_active_positions": mx,
            "daily_open_count": daily_opens,
            "daily_position_limit": daily_pos_lim,
            "daily_risk_used_pct": daily_risk_used,
            "daily_trading_risk_pct_limit": daily_risk_lim,
        }

        if daily:
            return {
                **base,
                "status": STATUS_DAILY,
                "status_label": STATUS_LABELS[STATUS_DAILY],
                "can_trade": False,
                "can_roll": False,
                "reason": "当日日冻结，禁止新开仓",
            }
        if daily_risk_limit_hit:
            return {
                **base,
                "status": STATUS_DAILY,
                "status_label": STATUS_LABELS[STATUS_DAILY],
                "can_trade": False,
                "can_roll": pos_limit,
                "reason": f"已达日交易风险上限 {daily_risk_used:.2f}%/{daily_risk_lim:.2f}%",
            }
        if daily_open_limit:
            return {
                **base,
                "status": STATUS_DAILY,
                "status_label": STATUS_LABELS[STATUS_DAILY],
                "can_trade": False,
                "can_roll": pos_limit,
                "reason": f"已达日持仓上限 {daily_opens}/{daily_pos_lim}",
            }
        if pos_limit:
            return {
                **base,
                "status": STATUS_FREEZE_POSITION,
                "status_label": STATUS_LABELS[STATUS_FREEZE_POSITION],
                "can_trade": False,
                "can_roll": True,
                "reason": f"已达仓位上限 {active}/{mx}",
            }
        return {
            **base,
            "status": STATUS_NORMAL,
            "status_label": STATUS_LABELS[STATUS_NORMAL],
            "can_trade": True,
            "can_roll": True,
            "reason": "可新开仓",
        }

    return _db_retry(_load)


def assert_can_open(
    conn,
    *,
    active_count: Optional[int] = None,
    equity: Optional[float] = None,
) -> Optional[str]:
    rs = get_risk_status(conn, active_count=active_count, equity=equity)
    if not rs.get("can_trade"):
        return rs.get("reason") or "当前不可开仓"
    return None