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

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

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

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:
    try:
        return max(0.0, float(os.getenv("RISK_COOLING_HOURS_MANUAL", "4")))
    except (TypeError, ValueError):
        return 4.0


def cooling_hours_manual_journal() -> float:
    try:
        return max(0.0, float(os.getenv("RISK_COOLING_HOURS_MANUAL_JOURNAL", "1")))
    except (TypeError, ValueError):
        return 1.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 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


def _db_retry(action: Callable[[], T], *, retries: int = 8, base_delay: float = 0.03) -> T:
    last: sqlite3.OperationalError | None = None
    for i in range(retries):
        try:
            return action()
        except sqlite3.OperationalError as exc:
            if "locked" not in str(exc).lower():
                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:
        return
    conn.execute(
        """CREATE TABLE IF NOT EXISTS account_risk_state (
            id INTEGER PRIMARY KEY CHECK (id = 1),
            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
        )"""
    )
    if not conn.execute("SELECT id 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 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, last_close_at_ms=?, updated_at=? WHERE id=1""",
            (td, count, close_ms, datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
        )
        return
    until = close_ms + int(cooling_hours_manual() * 3600 * 1000)
    conn.execute(
        """UPDATE account_risk_state SET trading_day=?, manual_close_count=?,
           daily_frozen=0, cooloff_until_ms=?, cooloff_hours=?, last_close_at_ms=?, updated_at=? WHERE id=1""",
        (td, count, until, int(cooling_hours_manual()), 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:
    """复盘手动平仓说明后，4h 冷静期降为 1h。"""
    if not risk_control_enabled():
        return
    ensure_account_risk_schema(conn)
    row = conn.execute("SELECT * FROM account_risk_state WHERE id=1").fetchone()
    if int(_row_get(row, "daily_frozen") or 0):
        return
    until = _row_get(row, "cooloff_until_ms")
    if not until:
        return
    now_ms = _now_ms(now)
    if int(until) <= now_ms:
        return
    last = int(_row_get(row, "last_close_at_ms") or now_ms)
    journal_ms = int(cooling_hours_manual_journal() * 3600 * 1000)
    new_until = max(now_ms, last + journal_ms)
    conn.execute(
        """UPDATE account_risk_state SET cooloff_until_ms=?, cooloff_hours=?, updated_at=? WHERE id=1""",
        (new_until, int(cooling_hours_manual_journal()), datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
    )


def get_risk_status(conn, *, now: Optional[datetime] = None, active_count: Optional[int] = None) -> dict:
    def _load() -> dict:
        ensure_account_risk_schema(conn)
        row = conn.execute("SELECT * FROM account_risk_state WHERE id=1").fetchone()
        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")
        active = count_active_trade_monitors(conn) if active_count is None else int(active_count)
        mx = max_active_positions()
        pos_limit = active >= mx

        if daily:
            return {
                "status": STATUS_DAILY,
                "status_label": STATUS_LABELS[STATUS_DAILY],
                "can_trade": False,
                "can_roll": False,
                "reason": "当日日冻结，禁止新开仓",
                "active_count": active,
                "max_active_positions": mx,
            }
        if until and int(until) > now_ms:
            rem = int((int(until) - now_ms) / 1000)
            hours = float(_row_get(row, "cooloff_hours") or cooling_hours_manual())
            st = STATUS_FREEZE_1H if hours <= cooling_hours_manual_journal() + 0.01 else STATUS_FREEZE_4H
            return {
                "status": st,
                "status_label": STATUS_LABELS[st],
                "can_trade": False,
                "can_roll": pos_limit,
                "reason": f"冷静期中，剩余约 {rem // 3600}h {(rem % 3600) // 60}m",
                "freeze_remaining_sec": rem,
                "active_count": active,
                "max_active_positions": mx,
            }
        if pos_limit:
            return {
                "status": STATUS_FREEZE_POSITION,
                "status_label": STATUS_LABELS[STATUS_FREEZE_POSITION],
                "can_trade": False,
                "can_roll": True,
                "reason": f"已达仓位上限 {active}/{mx}",
                "active_count": active,
                "max_active_positions": mx,
            }
        return {
            "status": STATUS_NORMAL,
            "status_label": STATUS_LABELS[STATUS_NORMAL],
            "can_trade": True,
            "can_roll": True,
            "reason": "可新开仓",
            "active_count": active,
            "max_active_positions": mx,
        }

    return _db_retry(_load)


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