crypto_monitor/daily_open_limit_lib.py

"""单日开仓次数：软提醒阈值 + 硬上限（四所实例共用）。"""
from __future__ import annotations

import os
from typing import Any, Optional


def parse_daily_open_alert_threshold(raw: Any = None, *, default: int = 5) -> int:
    """AI 克制提醒阈值；至少 1。"""
    try:
        v = int(raw if raw is not None and str(raw).strip() != "" else default)
    except (TypeError, ValueError):
        v = default
    return max(1, v)


def parse_daily_open_hard_limit(raw: Any = None, *, default: int = 0) -> int:
    """硬上限；0 表示不启用。至少 0。"""
    try:
        v = int(raw if raw is not None and str(raw).strip() != "" else default)
    except (TypeError, ValueError):
        v = default
    return max(0, v)


def load_daily_open_limits_from_env(
    env: Optional[dict[str, str]] = None,
) -> tuple[int, int]:
    """从环境变量读取 (alert_threshold, hard_limit)。"""
    src = env if env is not None else os.environ
    alert = parse_daily_open_alert_threshold(src.get("DAILY_OPEN_ALERT_THRESHOLD"))
    hard = parse_daily_open_hard_limit(src.get("DAILY_OPEN_HARD_LIMIT"))
    return alert, hard


def count_opens_for_trading_day(conn, trading_day: str) -> int:
    """本交易日已成功写入 order_monitors 的开仓次数。"""
    td = (trading_day or "").strip()
    if not td:
        return 0
    row = conn.execute(
        "SELECT COUNT(*) FROM order_monitors WHERE session_date=?",
        (td,),
    ).fetchone()
    return int(row[0] if row else 0)


def daily_open_hard_limit_blocks(opens_today: int, hard_limit: int) -> bool:
    return int(hard_limit) > 0 and int(opens_today) >= int(hard_limit)


def hard_limit_block_reason(opens_today: int, hard_limit: int, reset_hour: int) -> str:
    return (
        f"本交易日开仓次数已达上限（{int(opens_today)}/{int(hard_limit)}），"
        f"次日北京时间 {int(reset_hour)}:00 后恢复"
    )


def check_daily_open_hard_limit(
    conn,
    trading_day: str,
    hard_limit: int,
    reset_hour: int,
) -> tuple[bool, str, int]:
    """返回 (允许继续开仓, 拒绝原因, 当日已开次数)。"""
    opens_today = count_opens_for_trading_day(conn, trading_day)
    if daily_open_hard_limit_blocks(opens_today, hard_limit):
        return False, hard_limit_block_reason(opens_today, hard_limit, reset_hour), opens_today
    return True, "", opens_today


def can_trade_new_open(
    *,
    time_allows: bool,
    active_count: int,
    max_active_positions: int,
    opens_today: int,
    hard_limit: int,
    extra_blocks: bool = False,
) -> bool:
    if extra_blocks:
        return False
    if not time_allows:
        return False
    if int(active_count) >= int(max_active_positions):
        return False
    if daily_open_hard_limit_blocks(opens_today, hard_limit):
        return False
    return True


def should_send_daily_open_alert(before: int, after: int, alert_threshold: int) -> bool:
    return int(before) < int(alert_threshold) <= int(after)


def build_daily_open_alert_prompt(
    trading_day: str,
    opens_after: int,
    alert_threshold: int,
    *,
    hard_limit: int = 0,
    detail_line: str = "",
) -> str:
    hard_txt = (
        f"硬上限 {hard_limit} 次（已达后将禁止新开仓直至下一交易日）。"
        if int(hard_limit) > 0
        else "未配置单日硬上限。"
    )
    extra = f" {detail_line}" if detail_line else ""
    return (
        f"用户在北京时间交易日 {trading_day} 已累计开仓 {opens_after} 次"
        f"（AI 提醒阈值 {alert_threshold}；{hard_txt}）"
        f"{extra}"
        f"用户自述“上头了”。请给克制提醒。"
    )


def format_daily_open_counter_line(
    opens_today: int,
    alert_threshold: int,
    hard_limit: int,
) -> str:
    if int(hard_limit) > 0:
        return (
            f"📅 当日开仓次数：{int(opens_today)} / 硬上限 {int(hard_limit)} 次"
            f"（AI 提醒阈值 {int(alert_threshold)}）"
        )
    return (
        f"📅 当日开仓次数：{int(opens_today)} / AI 提醒阈值 {int(alert_threshold)} 次"
    )


def format_daily_open_summary_short(
    opens_today: int,
    alert_threshold: int,
    hard_limit: int,
) -> str:
    if int(hard_limit) > 0:
        return f"本交易日累计开仓：{int(opens_today)}（硬上限 {int(hard_limit)}，提醒 {int(alert_threshold)}）"
    return f"本交易日累计开仓：{int(opens_today)}（提醒阈值 {int(alert_threshold)}）"