Add personal license agreement and rename product section to tradable symbols.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
+129
-124
@@ -1,124 +1,129 @@
|
||||
"""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,
|
||||
}
|
||||
# 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,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user