dekun
130 lines
4.8 KiB
Python
130 lines
4.8 KiB
Python
# 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"})
|
|
|
|
|
|
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] = ""
|
|
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,
|
|
}
|