auto_trade_sys/trading_system/symbol_policy.py

"""
交易对分层策略：
- 正常：按原策略处理
- reduced：缩小仓位，并在策略侧提高信号门槛
- blocked：直接禁止自动开仓
"""
import logging
import re
from typing import Any, Dict, Set

logger = logging.getLogger(__name__)


def auto_trade_whitelist_allows(symbol: str, whitelist_raw: Any) -> bool:
    """
    自动交易白名单（窄宇宙）。

    - 当 AUTO_TRADE_SYMBOL_WHITELIST 为空/未配置时：不限制，与历史行为一致。
    - 非空时：仅名单内的合约允许自动下单；仍会扫描并生成推荐（由 strategy 层在自动分支前拦截）。
    """
    sym = (symbol or "").strip().upper()
    if not sym:
        return False
    allowed = _parse_symbol_list(whitelist_raw)
    if not allowed:
        return True
    return sym in allowed


def _parse_symbol_list(raw: Any) -> Set[str]:
    """兼容 string/list/dict 三种配置格式，统一转成大写 symbol 集合。"""
    if raw is None:
        return set()

    if isinstance(raw, dict):
        if "symbols" in raw:
            return _parse_symbol_list(raw.get("symbols"))
        if "items" in raw:
            return _parse_symbol_list(raw.get("items"))
        return set()

    if isinstance(raw, str):
        parts = re.split(r"[\s,;，\n\r\t]+", raw)
        return {p.strip().upper() for p in parts if p and p.strip()}

    if isinstance(raw, list):
        items: Set[str] = set()
        for item in raw:
            if isinstance(item, str):
                sym = item.strip()
            elif isinstance(item, dict):
                sym = str(item.get("symbol") or item.get("value") or "").strip()
            else:
                sym = ""
            if sym:
                items.add(sym.upper())
        return items

    return set()


def _safe_float(value: Any, default: float) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _safe_int(value: Any, default: int) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def resolve_symbol_trading_policy(symbol: str) -> Dict[str, Any]:
    """
    合并手动名单与统计名单，返回统一的交易对分层结果。

    返回字段：
    - blocked: 是否禁止自动开仓
    - reduced: 是否进入减仓观察名单
    - position_factor: reduced 时建议的仓位系数
    - signal_boost: reduced 时建议提高的最小信号门槛
    - sources: 命中的来源，便于日志定位
    """
    policy = {
        "blocked": False,
        "reduced": False,
        "position_factor": 1.0,
        "signal_boost": 0,
        "sources": [],
    }
    symbol = (symbol or "").strip().upper()
    if not symbol:
        return policy

    try:
        from config_manager import GlobalStrategyConfigManager

        mgr = GlobalStrategyConfigManager()
        stats_symbol = mgr.get("STATS_SYMBOL_FILTERS", {}) or {}

        manual_blocked = _parse_symbol_list(mgr.get("MANUAL_BLOCKED_SYMBOLS", ""))
        manual_reduced = _parse_symbol_list(mgr.get("MANUAL_REDUCED_SYMBOLS", ""))

        manual_position_factor = _safe_float(
            mgr.get("MANUAL_REDUCED_SYMBOL_POSITION_FACTOR", 0.5), 0.5
        )
        manual_position_factor = max(0.05, min(manual_position_factor, 1.0))

        manual_signal_boost = _safe_int(
            mgr.get("MANUAL_REDUCED_SYMBOL_SIGNAL_BOOST", 1), 1
        )
        manual_signal_boost = max(0, min(manual_signal_boost, 5))

        if symbol in manual_blocked:
            policy["blocked"] = True
            policy["sources"].append("manual_blocked")

        if symbol in manual_reduced:
            policy["reduced"] = True
            policy["position_factor"] = min(policy["position_factor"], manual_position_factor)
            policy["signal_boost"] = max(policy["signal_boost"], manual_signal_boost)
            policy["sources"].append("manual_reduced")

        for item in (stats_symbol.get("blacklist") or []):
            if (item.get("symbol") or "").strip().upper() != symbol:
                continue
            mode = (item.get("mode") or "reduced").strip().lower()
            if mode == "hard":
                policy["blocked"] = True
                policy["sources"].append("stats_hard")
            else:
                policy["reduced"] = True
                policy["position_factor"] = min(policy["position_factor"], 0.5)
                policy["signal_boost"] = max(policy["signal_boost"], 1)
                policy["sources"].append("stats_reduced")
            break
    except Exception as e:
        logger.debug(f"{symbol} 读取交易对分层策略失败: {e}")

    if policy["blocked"]:
        policy["reduced"] = False
        policy["position_factor"] = 1.0
        policy["signal_boost"] = 0

    # 去重并保留顺序，便于日志阅读。
    policy["sources"] = list(dict.fromkeys(policy["sources"]))
    return policy