qihuo/modules/keys/routes.py

# Copyright (c) 2025-2026 马建军. All rights reserved.
"""HTTP routes for keys 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

    @app.route("/api/key_prices")
    @login_required
    def api_key_prices():
        """关键位监控列表：批量现价与距上/下沿距离。"""
        conn = get_db()
        rows = conn.execute(
            "SELECT id, symbol, market_code, sina_code, upper, lower "
            "FROM key_monitors WHERE status='active' OR status IS NULL"
        ).fetchall()
        conn.close()
        out = []
        for r in rows:
            sym = r["symbol"]
            market = r["market_code"] or ""
            sina = r["sina_code"] or ""
            upper = float(r["upper"])
            lower = float(r["lower"])
            price = fetch_price(sym, market, sina)
            dist_upper = None
            dist_lower = None
            if price is not None:
                dist_upper = round(upper - price, 2)
                dist_lower = round(price - lower, 2)
            out.append({
                "id": r["id"],
                "price": price,
                "dist_upper": dist_upper,
                "dist_lower": dist_lower,
            })
        return jsonify(out)
    @app.route("/keys")
    @login_required
    def keys():
        from modules.keys.key_monitor_lib import key_monitor_periods

        conn = get_db()
        key_list = conn.execute(
            "SELECT * FROM key_monitors WHERE status='active' OR status IS NULL ORDER BY id DESC"
        ).fetchall()
        history = conn.execute(
            "SELECT * FROM key_monitors WHERE status='archived' ORDER BY archived_at DESC LIMIT 100"
        ).fetchall()
        conn.close()
        return render_template(
            "keys.html",
            keys=key_list,
            history=history,
            key_periods=key_monitor_periods(),
        )



    @app.route("/add_key", methods=["POST"])
    @login_required
    def add_key():
        d = request.form
        symbol = d.get("symbol", "").strip()
        symbol_name = d.get("symbol_name", "").strip()
        market_code = d.get("market_code", "").strip()
        sina_code = d.get("sina_code", "").strip()
        monitor_type = (d.get("type") or "").strip()
        if not symbol or not market_code:
            flash("请从下拉列表选择品种（同花顺合约代码）")
            return redirect(url_for("keys"))
        try:
            upper = float(d.get("upper") or 0)
            lower = float(d.get("lower") or 0)
        except (TypeError, ValueError):
            flash("上沿/下沿价格无效")
            return redirect(url_for("keys"))
        if upper <= lower:
            flash("上沿必须大于下沿")
            return redirect(url_for("keys"))

        trade_mode = (d.get("trade_mode") or "顺势").strip()
        if trade_mode not in ("顺势", "反转"):
            trade_mode = "顺势"
        try:
            risk_reward = float(d.get("risk_reward") or 2)
        except (TypeError, ValueError):
            risk_reward = 2.0
        risk_reward = max(0.5, min(10.0, risk_reward))
        trailing_be = 1 if d.get("trailing_be") else 0
        if trailing_be and risk_reward < 3:
            risk_reward = 3.0

        from modules.keys.key_monitor_lib import normalize_bar_period

        bar_period = normalize_bar_period(d.get("bar_period") or "5m")
        direction = (d.get("direction") or "").strip().lower()
        if monitor_type == "箱体突破":
            if direction not in ("long", "short"):
                flash("箱体突破须选择上方向（做多/做空）")
                return redirect(url_for("keys"))
        else:
            direction = ""

        conn = get_db()
        conn.execute(
            """INSERT INTO key_monitors
               (symbol, symbol_name, market_code, sina_code, monitor_type, direction,
                upper, lower, trade_mode, risk_reward, trailing_be, bar_period)
               VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
            (
                symbol, symbol_name, market_code, sina_code, monitor_type, direction,
                upper, lower, trade_mode, risk_reward, trailing_be, bar_period,
            ),
        )
        conn.commit()
        conn.close()
        flash("关键位监控已添加")
        return redirect(url_for("keys"))


    @app.route("/add_position", methods=["POST"])
    @login_required
    def add_position():
        flash("持仓由策略交易或 CTP 自动同步，无需手工录入")
        return redirect(url_for("positions"))
    @app.route("/del_key/<int:pid>")
    @login_required
    def del_key(pid):
        conn = get_db()
        conn.execute(
            "UPDATE key_monitors SET status='archived', archived_at=? WHERE id=?",
            (datetime.now(TZ).isoformat(), pid),
        )
        conn.commit()
        conn.close()
        flash("已移入监控历史")
        return redirect(url_for("keys"))