dekun
a1abe159fa
Add deploy/manage.sh bootstrap for git.bz121.com/dekun/crypto_okx and point docs at this repo. Co-authored-by: Cursor <cursoragent@cursor.com>
53 lines
1.3 KiB
Python
53 lines
1.3 KiB
Python
"""交易模式: sim | live, 持久化到 app_runtime_settings."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Callable
|
|
|
|
from lib.instance.runtime_settings_lib import runtime_get, runtime_set, with_db
|
|
|
|
TRADING_MODE_KEY = "trading.mode"
|
|
MODE_SIM = "sim"
|
|
MODE_LIVE = "live"
|
|
VALID_MODES = (MODE_SIM, MODE_LIVE)
|
|
|
|
|
|
def default_trading_mode() -> str:
|
|
raw = (os.getenv("SIM_DEFAULT_MODE") or "").strip().lower()
|
|
if raw in VALID_MODES:
|
|
return raw
|
|
return MODE_SIM
|
|
|
|
|
|
def normalize_mode(mode: str | None) -> str:
|
|
m = (mode or "").strip().lower()
|
|
if m in VALID_MODES:
|
|
return m
|
|
raise ValueError("mode 须为 sim 或 live")
|
|
|
|
|
|
def get_trading_mode(get_db: Callable) -> str:
|
|
def _read(conn):
|
|
v = runtime_get(conn, TRADING_MODE_KEY)
|
|
if v is None or str(v).strip() == "":
|
|
return default_trading_mode()
|
|
m = str(v).strip().lower()
|
|
return m if m in VALID_MODES else default_trading_mode()
|
|
|
|
return with_db(get_db, _read)
|
|
|
|
|
|
def set_trading_mode(get_db: Callable, mode: str) -> str:
|
|
m = normalize_mode(mode)
|
|
|
|
def _write(conn):
|
|
runtime_set(conn, TRADING_MODE_KEY, m)
|
|
return m
|
|
|
|
return with_db(get_db, _write)
|
|
|
|
|
|
def is_sim_mode(get_db: Callable) -> bool:
|
|
return get_trading_mode(get_db) == MODE_SIM
|