qihuo/ai_worker.py

# Copyright (c) 2025-2026 马建军. All rights reserved.
# 专有软件 — 未经授权禁止复制、传播、转售。
# 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。
# 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md

"""AI 后台：开仓/平仓分析、日终持仓报告。"""
from __future__ import annotations

import json
import logging
import threading
from datetime import datetime
from typing import Callable, Optional
from zoneinfo import ZoneInfo

logger = logging.getLogger(__name__)
TZ = ZoneInfo("Asia/Shanghai")
DAILY_REPORT_KEY = "ai_daily_report_last_date"


def schedule_ai_event_analysis(
    *,
    db_path: str,
    get_setting_fn: Callable[[str, str], str],
    kind: str,
    title: str,
    payload: dict,
    send_wechat_fn: Callable[[str], None] | None = None,
) -> None:
    """后台线程：调用 AI 并写入 ai_messages。"""
    if not (get_setting_fn("ai_enabled", "0") or "0").strip() in ("1", "true", "yes"):
        return

    def _run() -> None:
        from ai_client import analyze_trading_event
        from ai_messages import insert_ai_message
        from db_conn import connect_db

        ok, content = analyze_trading_event(
            get_setting=get_setting_fn,
            event_kind=kind,
            payload=payload,
        )
        if not ok:
            content = f"⚠ {content}"
        try:
            conn = connect_db(db_path)
            try:
                insert_ai_message(
                    conn,
                    kind=kind,
                    title=title,
                    content=content,
                    meta=payload,
                )
                conn.commit()
            finally:
                conn.close()
            if send_wechat_fn and ok:
                send_wechat_fn(f"🤖 AI 分析 · {title}\n\n{content[:1800]}")
        except Exception as exc:
            logger.warning("AI event analysis failed: %s", exc)

    threading.Thread(target=_run, daemon=True, name="ai-event").start()


def _today_trading_summary(conn, day: str) -> dict:
    rows = conn.execute(
        """SELECT symbol, symbol_name, direction, pnl_net, result, close_time
           FROM trade_logs WHERE close_time LIKE ? ORDER BY id ASC""",
        (f"{day}%",),
    ).fetchall()
    wins = losses = 0
    pnl_sum = 0.0
    trades = []
    for r in rows:
        pnl = float(r["pnl_net"] or 0)
        pnl_sum += pnl
        if pnl >= 0:
            wins += 1
        else:
            losses += 1
        trades.append(dict(r))
    positions = conn.execute(
        """SELECT symbol, symbol_name, direction, lots, entry_price, stop_loss, take_profit, monitor_type
           FROM trade_order_monitors WHERE status='active'"""
    ).fetchall()
    return {
        "date": day,
        "trade_count": len(trades),
        "wins": wins,
        "losses": losses,
        "pnl_net_total": round(pnl_sum, 2),
        "trades": trades[:20],
        "active_positions": [dict(p) for p in positions],
    }


def maybe_run_daily_ai_report(
    *,
    db_path: str,
    get_setting_fn: Callable[[str, str], str],
    set_setting_fn: Callable[[str, str], None],
    send_wechat_fn: Callable[[str], None] | None = None,
) -> None:
    if not (get_setting_fn("ai_enabled", "0") or "0").strip() in ("1", "true", "yes"):
        return
    if (get_setting_fn("ai_daily_report_enabled", "1") or "1").strip() not in ("1", "true", "yes"):
        return
    now = datetime.now(TZ)
    day = now.strftime("%Y-%m-%d")
    if get_setting_fn(DAILY_REPORT_KEY, "") == day:
        return
    try:
        hour = int(float(get_setting_fn("ai_daily_report_hour", "15") or 15))
        minute = int(float(get_setting_fn("ai_daily_report_minute", "5") or 5))
    except (TypeError, ValueError):
        hour, minute = 15, 5
    if (now.hour, now.minute) < (hour, minute):
        return

    from ai_client import analyze_trading_event
    from ai_messages import insert_ai_message
    from db_conn import connect_db

    try:
        conn = connect_db(db_path)
        try:
            summary = _today_trading_summary(conn, day)
            ok, content = analyze_trading_event(
                get_setting=get_setting_fn,
                event_kind="daily_report",
                payload=summary,
            )
            title = f"{day} 日终持仓与交易报告"
            if not ok:
                content = f"⚠ {content}"
            insert_ai_message(conn, kind="daily_report", title=title, content=content, meta=summary)
            conn.commit()
            set_setting_fn(DAILY_REPORT_KEY, day)
            if send_wechat_fn and ok:
                send_wechat_fn(f"🤖 {title}\n\n{content[:1800]}")
        finally:
            conn.close()
    except Exception as exc:
        logger.warning("AI daily report failed: %s", exc)


def start_ai_worker(
    *,
    db_path: str,
    get_setting_fn: Callable[[str, str], str],
    set_setting_fn: Callable[[str, str], None],
    send_wechat_fn: Callable[[str], None] | None = None,
    interval_sec: int = 60,
) -> None:
    import time

    def _loop() -> None:
        time.sleep(30)
        while True:
            try:
                maybe_run_daily_ai_report(
                    db_path=db_path,
                    get_setting_fn=get_setting_fn,
                    set_setting_fn=set_setting_fn,
                    send_wechat_fn=send_wechat_fn,
                )
            except Exception as exc:
                logger.debug("ai worker: %s", exc)
            time.sleep(max(30, interval_sec))

    threading.Thread(target=_loop, daemon=True, name="ai-worker").start()