qihuo/ctp_settings.py

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

"""CTP / SimNow 配置：系统设置优先，.env 作兜底。"""
from __future__ import annotations

import os
from typing import Any, Callable

# (db_key, env_key, vnpy字段名, 默认值)
SIMNOW_FIELDS: tuple[tuple[str, str, str, str], ...] = (
    ("simnow_user", "SIMNOW_USER", "用户名", ""),
    ("simnow_password", "SIMNOW_PASSWORD", "密码", ""),
    ("simnow_broker_id", "SIMNOW_BROKER_ID", "经纪商代码", "9999"),
    ("simnow_td_address", "SIMNOW_TD_ADDRESS", "交易服务器", "tcp://180.168.146.187:10201"),
    ("simnow_md_address", "SIMNOW_MD_ADDRESS", "行情服务器", "tcp://180.168.146.187:10211"),
    ("simnow_app_id", "SIMNOW_APP_ID", "产品名称", "simnow_client_test"),
    ("simnow_auth_code", "SIMNOW_AUTH_CODE", "授权编码", "0000000000000000"),
    ("simnow_env", "SIMNOW_ENV", "柜台环境", "实盘"),
)

LIVE_FIELDS: tuple[tuple[str, str, str, str], ...] = (
    ("ctp_live_user", "CTP_LIVE_USER", "用户名", ""),
    ("ctp_live_password", "CTP_LIVE_PASSWORD", "密码", ""),
    ("ctp_live_broker_id", "CTP_LIVE_BROKER_ID", "经纪商代码", ""),
    ("ctp_live_td_address", "CTP_LIVE_TD_ADDRESS", "交易服务器", ""),
    ("ctp_live_md_address", "CTP_LIVE_MD_ADDRESS", "行情服务器", ""),
    ("ctp_live_app_id", "CTP_LIVE_APP_ID", "产品名称", ""),
    ("ctp_live_auth_code", "CTP_LIVE_AUTH_CODE", "授权编码", ""),
    ("ctp_live_env", "CTP_LIVE_ENV", "柜台环境", "实盘"),
)

PASSWORD_DB_KEYS = frozenset({"simnow_password", "ctp_live_password"})

CTP_AUTO_CONNECT_KEY = "ctp_auto_connect"
CTP_DISABLED_HINT = "CTP 自动连接已关闭（非交易时段建议关闭，避免反复连接失败）"


def is_ctp_auto_connect_enabled(get_setting=None) -> bool:
    """系统设置：是否允许 CTP 连接（含自动重连、盘前连接、手动连接）。"""
    if get_setting is None:
        from fee_specs import get_setting as _gs

        get_setting = _gs
    val = (get_setting(CTP_AUTO_CONNECT_KEY, "1") or "1").strip().lower()
    return val in ("1", "true", "yes", "on")


def save_ctp_auto_connect(form: Any, set_setting: Callable[[str, str], None]) -> bool:
    enabled = (form.get("ctp_auto_connect") or "").strip().lower() in (
        "1",
        "on",
        "true",
        "yes",
    )
    set_setting(CTP_AUTO_CONNECT_KEY, "1" if enabled else "0")
    return enabled


def _get_db_setting(key: str, default: str = "") -> str:
    from fee_specs import get_setting

    return (get_setting(key, default) or default).strip()


def resolve_ctp_value(db_key: str, env_key: str, default: str = "") -> str:
    v = _get_db_setting(db_key, "")
    if v:
        return v
    return (os.getenv(env_key) or default).strip()


def _build_setting_dict(fields: tuple[tuple[str, str, str, str], ...]) -> dict[str, str]:
    out: dict[str, str] = {}
    for db_key, env_key, vnpy_key, default in fields:
        out[vnpy_key] = resolve_ctp_value(db_key, env_key, default)
    return out


def simnow_setting_dict() -> dict[str, str]:
    return _build_setting_dict(SIMNOW_FIELDS)


def live_setting_dict() -> dict[str, str]:
    return _build_setting_dict(LIVE_FIELDS)


def seed_ctp_settings_from_env(set_setting: Callable[[str, str], None]) -> None:
    """首次启动：将 .env 中已有 CTP 配置写入 settings 表。"""
    for db_key, env_key, _, _ in (*SIMNOW_FIELDS, *LIVE_FIELDS):
        if _get_db_setting(db_key, ""):
            continue
        env_val = (os.getenv(env_key) or "").strip()
        if env_val:
            set_setting(db_key, env_val)


def get_ctp_settings_for_ui() -> dict[str, Any]:
    ui: dict[str, Any] = {}
    for db_key, env_key, _, default in SIMNOW_FIELDS:
        ui[db_key] = resolve_ctp_value(db_key, env_key, default)
        if db_key in PASSWORD_DB_KEYS:
            ui[f"{db_key}_set"] = bool(ui[db_key])
            ui[db_key] = ""
    for db_key, env_key, _, default in LIVE_FIELDS:
        ui[db_key] = resolve_ctp_value(db_key, env_key, default)
        if db_key in PASSWORD_DB_KEYS:
            ui[f"{db_key}_set"] = bool(ui[db_key])
            ui[db_key] = ""
    ui["ctp_auto_connect"] = is_ctp_auto_connect_enabled()
    return ui


def save_ctp_settings_from_form(
    form: Any,
    set_setting: Callable[[str, str], None],
) -> dict[str, Any]:
    """保存 CTP 配置；密码留空表示不修改。返回摘要供页面提示。"""
    passwords_updated: list[str] = []
    passwords_submitted_empty: list[str] = []

    for db_key, _, _, default in SIMNOW_FIELDS:
        if db_key in PASSWORD_DB_KEYS:
            raw = form.get(db_key)
            val = (raw or "").strip()
            if val:
                set_setting(db_key, val)
                passwords_updated.append(db_key)
            else:
                passwords_submitted_empty.append(db_key)
            continue
        val = (form.get(db_key) or "").strip()
        set_setting(db_key, val or default)

    for db_key, _, _, default in LIVE_FIELDS:
        if db_key in PASSWORD_DB_KEYS:
            raw = form.get(db_key)
            val = (raw or "").strip()
            if val:
                set_setting(db_key, val)
                passwords_updated.append(db_key)
            else:
                passwords_submitted_empty.append(db_key)
            continue
        val = (form.get(db_key) or "").strip()
        if default or val:
            set_setting(db_key, val or default)

    return {
        "passwords_updated": passwords_updated,
        "passwords_submitted_empty": passwords_submitted_empty,
    }