# Copyright (c) 2025-2026 马建军. All rights reserved. """HTTP routes for settings module.""" from __future__ import annotations from datetime import date, datetime from flask import ( Response, flash, jsonify, redirect, render_template, request, send_file, session, stream_with_context, url_for, ) def register(deps) -> None: app = deps.app login_required = deps.login_required require_nav = deps.require_nav get_db = deps.get_db get_setting = deps.get_setting set_setting = deps.set_setting fetch_price = deps.fetch_price send_wechat_msg = deps.send_wechat_msg touch_stats_cache = deps.touch_stats_cache get_stats_data = deps.get_stats_data build_market_quote_payload = deps.build_market_quote_payload today_str = deps.today_str expire_old_plans = deps.expire_old_plans TZ = deps.tz DB_PATH = deps.db_path UPLOAD_DIR = deps.upload_dir OPEN_TYPES = deps.open_types EXIT_TRIGGERS = deps.exit_triggers BEHAVIOR_TAGS = deps.behavior_tags KLINE_PERIODS = deps.kline_periods KLINE_CUTOFFS = deps.kline_cutoffs calc_holding_duration = deps.calc_holding_duration holding_to_minutes = deps.holding_to_minutes classify_close_result = deps.classify_close_result calc_rr_ratio = deps.calc_rr_ratio calc_theoretical_pnl = deps.calc_theoretical_pnl parse_review_date_filter = deps.parse_review_date_filter _trading_mode = deps.trading_mode _ua_is_phone = deps.ua_is_phone _static_asset_v = deps.static_asset_v from modules.settings.nav_settings import NAV_TOGGLES, get_nav_items, save_nav_items from modules.settings.admin_settings import save_admin_credentials from modules.backup.db_backup import ( backup_dir, backup_in_progress, default_restore_dir, get_backup_last_at, list_backups, schedule_backup, ) from modules.market.market import get_quote_source_label from modules.trading.product_recommend import small_account_margin_recommendations @app.route("/settings", methods=["GET", "POST"]) @login_required def settings(): if request.method == "POST": action = request.form.get("action") if action == "backup_now": ok, msg = schedule_backup( get_setting=get_setting, set_setting=set_setting, include_uploads=True, ) flash(msg if ok else msg) elif action == "backup_config": auto = request.form.get("backup_auto_enabled") == "1" set_setting("backup_auto_enabled", "1" if auto else "0") try: hour = int(request.form.get("backup_auto_hour", "3") or 3) set_setting("backup_auto_hour", str(max(0, min(23, hour)))) except ValueError: flash("自动备份小时无效") return redirect(url_for("settings")) try: keep = int(request.form.get("backup_keep_count", "30") or 30) set_setting("backup_keep_count", str(max(5, min(200, keep)))) except ValueError: flash("保留份数无效") return redirect(url_for("settings")) flash("备份策略已保存") elif action == "wechat": webhook = request.form.get("wechat_webhook", "").strip() set_setting("wechat_webhook", webhook) flash("企业微信配置已保存") elif action == "ai": set_setting("ai_enabled", "1" if request.form.get("ai_enabled") else "0") provider = (request.form.get("ai_provider") or "ollama").strip().lower() if provider not in ("ollama", "openai"): provider = "ollama" set_setting("ai_provider", provider) set_setting("ai_ollama_base_url", (request.form.get("ai_ollama_base_url") or "").strip()) set_setting("ai_ollama_model", (request.form.get("ai_ollama_model") or "").strip()) set_setting("ai_openai_base_url", (request.form.get("ai_openai_base_url") or "").strip()) key = (request.form.get("ai_openai_api_key") or "").strip() if key: set_setting("ai_openai_api_key", key) set_setting("ai_openai_model", (request.form.get("ai_openai_model") or "").strip()) set_setting("ai_daily_report_enabled", "1" if request.form.get("ai_daily_report_enabled") else "0") try: set_setting("ai_daily_report_hour", str(max(0, min(23, int(request.form.get("ai_daily_report_hour", "15") or 15))))) except ValueError: pass try: set_setting("ai_daily_report_minute", str(max(0, min(59, int(request.form.get("ai_daily_report_minute", "5") or 5))))) except ValueError: pass flash("AI 配置已保存") elif action == "trading": mode = request.form.get("trading_mode", "simulation").strip() if mode not in ("simulation", "live"): mode = "simulation" sizing = request.form.get("position_sizing_mode", "fixed").strip() if sizing == "risk": sizing = "amount" if sizing not in ("fixed", "amount"): sizing = "fixed" set_setting("trading_mode", mode) set_setting("position_sizing_mode", sizing) try: fl = int(float(request.form.get("fixed_lots", "1") or 1)) set_setting("fixed_lots", str(max(1, fl))) except ValueError: flash("固定手数无效") return redirect(url_for("settings")) try: fa = float(request.form.get("fixed_amount", "5000") or 5000) set_setting("fixed_amount", str(max(1.0, fa))) except ValueError: flash("固定金额无效") return redirect(url_for("settings")) try: rp = float(request.form.get("risk_percent", "1") or 1) set_setting("risk_percent", str(max(0.1, min(100.0, rp)))) except ValueError: pass try: mp = float(request.form.get("max_margin_pct", "30") or 30) set_setting("max_margin_pct", str(max(1.0, min(100.0, mp)))) except ValueError: flash("保证金比例无效") return redirect(url_for("settings")) try: rmp = float(request.form.get("roll_max_margin_pct", "50") or 50) set_setting("roll_max_margin_pct", str(max(1.0, min(100.0, rmp)))) except ValueError: flash("滚仓保证金比例无效") return redirect(url_for("settings")) try: tb = int(float(request.form.get("trailing_be_tick_buffer", "2") or 2)) set_setting("trailing_be_tick_buffer", str(max(1, min(20, tb)))) except ValueError: flash("移动保本缓冲无效") return redirect(url_for("settings")) try: pt = int(float(request.form.get("pending_order_timeout_min", "5") or 5)) set_setting("pending_order_timeout_min", str(max(1, min(60, pt)))) except ValueError: flash("挂单超时无效") return redirect(url_for("settings")) flash("交易模式已保存") elif action == "ctp": from modules.ctp.ctp_settings import save_ctp_auto_connect, is_ctp_auto_connect_enabled from modules.ctp.ctp_settings import save_ctp_settings_from_form from modules.ctp.vnpy_bridge import ctp_disconnect was_enabled = is_ctp_auto_connect_enabled(get_setting) auto_enabled = save_ctp_auto_connect(request.form, set_setting) save_result = save_ctp_settings_from_form(request.form, set_setting) if not auto_enabled: ctp_disconnect(set_disabled_hint=True) elif not was_enabled and auto_enabled: try: from modules.ctp.vnpy_bridge import get_bridge from modules.core.trading_context import get_trading_mode mode = get_trading_mode(get_setting) get_bridge().reconnect_after_settings_saved(mode) except Exception as exc: app.logger.debug("CTP connect after enable auto: %s", exc) pwd_updated = save_result.get("passwords_updated") or [] pwd_empty = save_result.get("passwords_submitted_empty") or [] simnow_pwd_len = len((request.form.get("simnow_password") or "").strip()) live_pwd_len = len((request.form.get("ctp_live_password") or "").strip()) print( f"CTP settings save: simnow_password_len={simnow_pwd_len} " f"live_password_len={live_pwd_len} updated={pwd_updated}", flush=True, ) app.logger.info( "CTP settings save: simnow_password_len=%s live_password_len=%s updated=%s", simnow_pwd_len, live_pwd_len, pwd_updated, ) if "simnow_password" in pwd_updated: pwd_note = f"SimNow 交易密码已更新（{simnow_pwd_len} 位）" elif "simnow_password" in pwd_empty: pwd_note = "SimNow 交易密码未改：提交为空，请在「交易密码」框手打后再保存" elif "ctp_live_password" in pwd_updated: pwd_note = "实盘交易密码已更新" elif "ctp_live_password" in pwd_empty: pwd_note = "实盘交易密码未改（提交为空）" else: pwd_note = "" if not auto_enabled: flash("CTP 配置已保存；自动连接已关闭，所有 CTP 连接已断开") return redirect(url_for("settings")) if not was_enabled: flash("CTP 配置已保存；自动连接已开启，正在连接…") return redirect(url_for("settings")) flash_msg = "CTP 配置已保存，正在使用新地址重连…" if pwd_note: flash_msg = f"CTP 配置已保存；{pwd_note}，正在重连…" try: from modules.ctp.vnpy_bridge import get_bridge from modules.core.trading_context import get_trading_mode b = get_bridge() if pwd_updated: b._clear_login_cooldown() mode = get_trading_mode(get_setting) info = b.reconnect_after_settings_saved(mode) if info.get("cooldown"): flash_msg = f"CTP 配置已保存；{pwd_note or '请稍后再连'}" elif not info.get("started") and info.get("connected"): flash_msg = f"CTP 配置已保存；{pwd_note or '当前连接正常'}" except Exception as exc: app.logger.warning("CTP reconnect after settings save: %s", exc) flash_msg = f"CTP 配置已保存；{pwd_note or '请稍后在持仓监控页重连'}" flash(flash_msg) elif action == "nav": items = {k: request.form.get(f"nav_{k}") == "on" for k in NAV_TOGGLES} save_nav_items(set_setting, items) flash("导航显示已保存") elif action == "password": ok, msg, _ = save_admin_credentials( username=request.form.get("admin_username", ""), old_password=request.form.get("old_password", ""), new_password=request.form.get("new_password", ""), new_password2=request.form.get("new_password2", ""), get_setting=get_setting, set_setting=set_setting, ) if ok and session.get("logged_in"): session["username"] = (request.form.get("admin_username") or "").strip() flash(msg) return redirect(url_for("settings")) webhook = get_setting("wechat_webhook") username = get_setting("admin_username") ctp_st = {} try: from modules.ctp.vnpy_bridge import ctp_status from modules.core.trading_context import get_trading_mode ctp_st = ctp_status(get_trading_mode(get_setting)) except Exception: pass from modules.ctp.ctp_settings import get_ctp_settings_for_ui, is_ctp_auto_connect_enabled from modules.trading.product_recommend import small_account_margin_recommendations return render_template( "settings.html", webhook=webhook, username=username, quote_label=get_quote_source_label(ctp_connected=bool(ctp_st.get("connected"))), ctp_status=ctp_st, ctp_cfg=get_ctp_settings_for_ui(), ctp_auto_connect=is_ctp_auto_connect_enabled(get_setting), trading_mode=get_setting("trading_mode", "simulation"), position_sizing_mode=get_setting("position_sizing_mode", "fixed"), fixed_lots=get_setting("fixed_lots", "1"), fixed_amount=get_setting("fixed_amount", "5000"), risk_percent=get_setting("risk_percent", "1"), max_margin_pct=get_setting("max_margin_pct", "30"), roll_max_margin_pct=get_setting("roll_max_margin_pct", "50"), small_account_margin_rec=small_account_margin_recommendations(), trailing_be_tick_buffer=get_setting("trailing_be_tick_buffer", "2"), pending_order_timeout_min=get_setting("pending_order_timeout_min", "5"), nav_items=get_nav_items(get_setting), nav_toggles=NAV_TOGGLES, backup_dir=str(backup_dir()), backup_last_at=get_backup_last_at(get_setting), backup_running=backup_in_progress(), backup_items=list_backups(), backup_auto_enabled=get_setting("backup_auto_enabled", "1") == "1", backup_auto_hour=get_setting("backup_auto_hour", "3"), backup_keep_count=get_setting("backup_keep_count", "30"), backup_restore_dir=default_restore_dir(), ai_enabled=get_setting("ai_enabled", "0") == "1", ai_provider=get_setting("ai_provider", "ollama"), ai_ollama_base_url=get_setting("ai_ollama_base_url", "http://127.0.0.1:11434"), ai_ollama_model=get_setting("ai_ollama_model", "qwen2.5:7b"), ai_openai_base_url=get_setting("ai_openai_base_url", "https://api.openai.com/v1"), ai_openai_api_key=get_setting("ai_openai_api_key", ""), ai_openai_model=get_setting("ai_openai_model", "gpt-4o-mini"), ai_daily_report_enabled=get_setting("ai_daily_report_enabled", "1") == "1", ai_daily_report_hour=get_setting("ai_daily_report_hour", "15"), ai_daily_report_minute=get_setting("ai_daily_report_minute", "5"), )