#!/usr/bin/env python3 """MoFin Dashboard - 莫荷持仓情报可视化系统""" import base64 import json import os import re import uuid import urllib.request from datetime import datetime from pathlib import Path import sys sys.path.insert(0, "/home/hmo/MoFin/scripts") sys.path.insert(0, "/home/hmo/MoFin") from flask import Flask, jsonify, send_from_directory, request import socket import time import sqlite3 SPECS_DIR = Path(__file__).parent / "specs" GATEWAY_TEMP = Path(__file__).parent / "gateway" / "temp" START_TIME = time.time() # ── Dashboard 监控服务列表 ── DASH_SERVICES = [ {"name": "mofin_api", "label": "MoFin API", "port": 8899, "host": "127.0.0.1", "type": "http", "check": "/api/portfolio", "layer": "核心服务", "critical": True}, {"name": "zhiwei_gateway", "label": "知微 Gateway", "port": 8643, "host": "127.0.0.1", "type": "http", "check": "/v1/health", "layer": "AI 网关", "critical": True}, {"name": "ejabberd", "label": "ejabberd XMPP", "port": 5222, "host": "127.0.0.1", "type": "tcp", "check": None, "layer": "通信层", "critical": True}, {"name": "mofin_db", "label": "MoFin 数据库", "port": 0, "host": "127.0.0.1", "type": "db", "check": "/home/hmo/web-dashboard/data/mofin.db", "layer": "数据层", "critical": True}, ] def _chk_tcp(host, port, timeout=3): try: s = socket.create_connection((host, port), timeout=timeout) s.close() return True except Exception: return False def _chk_http(host, port, path, timeout=3): try: url = f"http://{host}:{port}{path}" urllib.request.urlopen(urllib.request.Request(url), timeout=timeout) return True except Exception: return False def _chk_db(db_path): try: conn = sqlite3.connect(db_path) conn.execute("SELECT 1") conn.close() return True except Exception: return False def _check_svc(svc): if svc["type"] == "tcp": return _chk_tcp(svc["host"], svc["port"]) elif svc["type"] == "http": return _chk_http(svc["host"], svc["port"], svc["check"]) elif svc["type"] == "db": return _chk_db(svc["check"]) return False # 提示词管理模块 from prompt_manager.dashboard_views import register_routes # MoFin 数据层（纯 DB，不再读 JSON） from mo_data import read_portfolio, read_decisions, read_watchlist from mofin_db import get_conn, write_holdings_batch, write_portfolio_summary, write_watchlist_stock, write_holding_strategy app = Flask(__name__, static_folder="static", static_url_path="") app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0 # 禁静态缓存：前端迭代频繁，防浏览器旧版残留 DATA_DIR = Path(__file__).parent / "data" UPLOAD_DIR = Path(__file__).parent / "uploads" # Hermes Gateway GATEWAY = "http://localhost:8642/v1/chat/completions" API_KEY = "hermes123" def _load_json(path, default=None): """仅用于非核心文件（reports, stocks, market 等）。portfolio/decisions/watchlist 已迁移到 DB。""" try: with open(path, encoding="utf-8") as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {} if default is None else default def _save_json(path, data): """仅用于非核心文件（reports, stocks, market 等）。portfolio/decisions/watchlist 已迁移到 DB。""" os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def _save_portfolio(data): """写入持仓数据到 DB。data 必须包含 holdings[] 和顶层 summary 字段。""" conn = get_conn() try: write_holdings_batch(conn, data.get('holdings', [])) write_portfolio_summary(conn, data) finally: conn.close() def _save_decision(code, name, data): """写入单条决策到 DB。""" conn = get_conn() try: write_holding_strategy(conn, code, name, data) finally: conn.close() def _save_watchlist(data): """写入自选股列表到 DB。""" conn = get_conn() for s in data.get('stocks', []): s.setdefault('currency', 'CNY') write_watchlist_stock(conn, s) conn.close() # ── API 路由 ────────────────────────────────────────── @app.route("/") def index(): return send_from_directory(app.static_folder, "index.html") @app.route("/api/watch") def get_watch(): """盯盘：所有有效策略（持仓+自选），服务端排序""" import sqlite3 conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") conn.row_factory = sqlite3.Row # 1) 所有 active 持仓策略 + 自选策略 rows = conn.execute(""" SELECT hs.code, hs.name, hs.decision_type, hs.timing_signal, hs.action, hs.position_advice, hs.tag, hs.entry_low, hs.entry_high, hs.stop_loss, hs.take_profit, hs.rr_ratio, hs.rr_low, hs.rr_high, hs.full_analysis, hs.reassessed_at, hs.rec_score, lp.price, lp.change_pct, h.shares, h.position_pct FROM holding_strategies hs LEFT JOIN live_prices lp ON hs.code = lp.code LEFT JOIN holdings h ON hs.code = h.code AND h.is_active = 1 WHERE hs.status='active' AND hs.decision_type IN ('持仓策略','自选策略') """).fetchall() conn.close() # 信号强度排序映射 signal_rank = { '买入': 1, '可买入': 2, '可加仓': 3, '止盈': 4, '卖出': 5, '关注': 6, '观望': 7, '持有': 8, '弱势持有': 9, '信号不充分': 10, } results = [] for r in rows: d = dict(r) # 分类 sort_group tag = d.get('tag') or '' if tag in ('current_recommend', 'active_manual'): d['sort_group'] = 0 # 推荐 elif d['decision_type'] == '持仓策略': d['sort_group'] = 1 # 持仓 else: d['sort_group'] = 2 # 自选 sig = d.get('timing_signal') or '' d['_sig_rank'] = signal_rank.get(sig, 99) # 持仓仓位（用于持仓组内排序） d['_pos'] = d.get('position_pct') or 0 d['_rr'] = d.get('rr_ratio') or 0 # 截断 full_analysis fa = d.get('full_analysis') or '' if len(fa) > 4000: fa = fa[:4000] + '\n...(已截断)' d['full_analysis'] = fa results.append(d) # ── 推荐操作精选层（2026-07-21 老爸：太多推荐=没有推荐）── # 候选 = tag 非空 且 72h 内有新鲜重评；按 RR 降序；贪心装入现金预算；最多 5 只。 # 落选者降回其自然分组（持仓/自选）。 import re as _re from datetime import datetime as _dt, timedelta as _td conn2 = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") try: _pr = conn2.execute("SELECT cash, total_assets FROM portfolio_summary WHERE id=1").fetchone() _cash = float(_pr[0] or 0) _total = float(_pr[1] or 0) finally: conn2.close() _budget_pct = (_cash / _total * 100) if _total > 0 else 0 _fresh_cutoff = _dt.now() - _td(hours=72) def _is_fresh(d): ra = d.get('reassessed_at') or '' if not ra: return False try: return _dt.fromisoformat(str(ra)[:19]) >= _fresh_cutoff except Exception: return False def _sugg_pct(d): # 从 position_advice 解析百分比（如 "8%（理由...)"），失败默认 8% m = _re.search(r'(\d+(?:\.\d+)?)\s*%', d.get('position_advice') or '') if m: try: v = float(m.group(1)) if 0 < v <= 30: return v except Exception: pass return 8.0 _SELL_SIGS = ("卖出", "止盈") _cands = [d for d in results if d['sort_group'] == 0 and _is_fresh(d)] # 卖出类永远可执行（释放现金，不占买入预算），排最前；买入类按 RR 降序 _sells = [d for d in _cands if (d.get('timing_signal') or '') in _SELL_SIGS] _buys = [d for d in _cands if (d.get('timing_signal') or '') not in _SELL_SIGS] _buys.sort(key=lambda x: (x.get('rec_score') or 0, x.get('rr_ratio') or 0), reverse=True) for d in _sells: d['rec_exec'] = True # 卖出不需要现金，永远可执行 d['suggested_position_pct'] = 0.0 # 买入：score≥60 + RR≥2.0 才有可执行资格（2026-07-27 老爸：五维评分替代纯RR） # 达标者贪心装入现金预算，预算外/不达标标记"排队"（不再降级隐藏） _cum = 0.0 # 弱信号不可执行 _WEAK_SIGNALS = ('信号不充分', '关注', '弱势持有', '观望', '持有', '') _TOP_N = 5 # 优中选优：推荐区只展示 Top 5（剩余排入自选区） for d in _buys: pct = _sugg_pct(d) d['suggested_position_pct'] = pct rr = d.get('rr_ratio') or 0 score = d.get('rec_score') or 0 sig_now = d.get('timing_signal') or '' # 仓位必须明确% _has_pos = bool(_re.search(r'(\d+(?:\.\d+)?)\s*%', d.get('position_advice') or '')) if sig_now in _WEAK_SIGNALS: d['rec_exec'] = False # 弱信号永远排队 elif not _has_pos: d['rec_exec'] = False # 无明确仓位，排队 elif score >= 60 and rr >= 2.0 and _cum + pct <= _budget_pct + 1e-9: d['rec_exec'] = True # 可执行（2026-07-24 老爸：门槛1.5→2.0，边缘推荐不算优） _cum += pct else: d['rec_exec'] = False # 排队（现金不足或RR不达标） # 落选（tag 但非新鲜）仍降回自然分组；新鲜者全部留在推荐区 for d in results: if d['sort_group'] == 0 and not _is_fresh(d): d['sort_group'] = 1 if d['decision_type'] == '持仓策略' else 2 # ── 优中选优（2026-07-27 老爸）：推荐区只展示 Top 5 买入，太多选不过来 ── # 卖出/止盈永远保留在推荐区；买入按评分降序，第6名起降入自选区 _rec_buys = [d for d in results if d['sort_group'] == 0 and (d.get('timing_signal') or '') not in _SELL_SIGS] _rec_buys.sort(key=lambda x: (x.get('rec_score') or 0, x.get('rr_ratio') or 0), reverse=True) for d in _rec_buys[_TOP_N:]: d['sort_group'] = 2 # 超额买入降入自选区 # 排序：group → signal_rank → group-internal (持仓按position_pct desc, 自选按rr desc) def skey(x): g = x['sort_group'] sr = x['_sig_rank'] # 同 signal 时持仓按仓位、自选按RR inner = x['_pos'] if g == 1 else x['_rr'] return (g, sr, -inner, x.get('code', '')) results.sort(key=skey) # 移除辅助排序键 for d in results: d.pop('_sig_rank', None) d.pop('_pos', None) d.pop('_rr', None) # 数据最新更新时间（price_monitor/live_prices） _data_time = "" try: _dtc = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") _r = _dtc.execute("SELECT MAX(updated_at) FROM live_prices").fetchone() if _r and _r[0]: _data_time = str(_r[0]) _dtc.close() except Exception: pass return json.dumps({"stocks": results, "count": len(results), "cash": _cash, "total_assets": _total, "budget_pct": round(_budget_pct, 2), "rec_used_pct": round(_cum, 2), "data_time": _data_time}, ensure_ascii=False) @app.route("/api/strategy_history/") def api_strategy_history(code): """某只股票最近 N 条策略记录（strategy_history 表）""" limit = min(int(request.args.get('limit', 3)), 20) import sqlite3 conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") conn.row_factory = sqlite3.Row try: rows = conn.execute(""" SELECT id, code, name, decision_type, strategy_type, full_analysis, action, timing_signal, entry_low, entry_high, stop_loss, take_profit, position_advice, rr_ratio, version, source_trigger, reassessed_at, snapshotted_at FROM strategy_history WHERE code=? ORDER BY snapshotted_at DESC LIMIT ? """, (code, limit)).fetchall() history = [dict(r) for r in rows] if not history: raise LookupError("history empty, fallback to current row") except Exception: # 表不存在或查询失败 → 降级为当前 holding_strategies 行 history = [] try: cur = conn.execute(""" SELECT code, name, decision_type, timing_signal, action, entry_low, entry_high, stop_loss, take_profit, rr_ratio, position_advice, full_analysis, reassessed_at FROM holding_strategies WHERE code=? AND status='active' """, (code,)).fetchone() if cur: d = dict(cur) d['version'] = 'current' d['snapshotted_at'] = d.get('reassessed_at', '') d['is_current'] = True history = [d] except Exception: history = [] finally: conn.close() return jsonify({"code": code, "count": len(history), "history": history}) _CRON_ID_MAP_CACHE = {"ts": 0, "map": {}} def _cron_name_to_id(): """从两个 profile 的 hermes jobs.json 构建 name->id 映射（60s 缓存）。""" import time as _t, glob as _g, json as _j now = _t.time() if now - _CRON_ID_MAP_CACHE["ts"] < 60: return _CRON_ID_MAP_CACHE["map"] m = {} for pj in _g.glob("/home/hmo/.hermes/profiles/*/cron/jobs.json"): try: with open(pj, encoding="utf-8") as f: jobs = _j.load(f) jobs = jobs if isinstance(jobs, list) else jobs.get("jobs", []) for j in jobs: jid, jname = str(j.get("id", "")), str(j.get("name", "")) if jid: m[jid] = jid if jname and jid: m[jname] = jid except Exception: pass _CRON_ID_MAP_CACHE["ts"] = now _CRON_ID_MAP_CACHE["map"] = m return m @app.route("/api/reports") def api_reports(): """历史报告列表，支持 ?cron=&script=<脚本名>&limit=N 过滤 匹配链：pipeline名→jobs.json解析为job id→文件名前缀 cron_{id}_ → pipeline名子串 → 脚本名(去.py)子串 → 报告title子串。 """ reports_dir = DATA_DIR / "reports" reports = [] if reports_dir.exists(): cron_key = (request.args.get("cron") or "").strip() script_key = (request.args.get("script") or "").strip() if script_key.endswith(".py"): script_key = script_key[:-3] limit = min(int(request.args.get("limit", 100)), 200) job_id = "" if cron_key: job_id = _cron_name_to_id().get(cron_key, "") for f in sorted(reports_dir.iterdir(), reverse=True): if f.suffix != ".json": continue data = _load_json(f) if (cron_key or script_key) else None if cron_key or script_key: stem = f.stem title = str(data.get("title", "")) # 命中链：job id 前缀 → 名/脚本子串(文件名) → pipeline名子串(标题) hit = False if job_id and stem.startswith(f"cron_{job_id}_"): hit = True elif cron_key and cron_key in stem: hit = True elif script_key and script_key in stem: hit = True elif cron_key and cron_key in title: hit = True if not hit: continue if data is None: data = _load_json(f) reports.append({ "id": f.stem, "title": data.get("title", f.stem), "type": data.get("type", "未知"), "created_at": data.get("created_at", ""), "summary": data.get("summary", ""), "cron": data.get("cron") or data.get("job") or "", }) if len(reports) >= limit: break return jsonify(reports) @app.route("/api/portfolio") def api_portfolio(): """持仓列表""" try: from mofin_db import get_conn, query_holdings, query_portfolio_summary conn = get_conn() holdings = query_holdings(conn) summary = query_portfolio_summary(conn) conn.close() if holdings: data = dict(summary) data["holdings"] = holdings return jsonify(data) except Exception: pass return jsonify({"error": "数据库查询失败"}), 500 @app.route("/api/watchlist") def api_watchlist(): """自选列表（从holding_strategies读取）""" try: import sqlite3 conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT hs.code, hs.name, hs.entry_low, hs.entry_high, hs.stop_loss, hs.take_profit, hs.timing_signal, hs.action, lp.price, lp.change_pct, hs.rr_ratio, hs.updated_at, hs.tech_snapshot, hs.sector_context, hs.full_analysis, hs.position_advice FROM holding_strategies hs LEFT JOIN live_prices lp ON hs.code = lp.code WHERE hs.status='active' AND hs.decision_type='自选策略' ORDER BY CASE WHEN hs.timing_signal IN ('买入','可买入','可加仓') THEN 0 WHEN hs.timing_signal IN ('关注') THEN 1 WHEN hs.timing_signal IN ('信号不充分') THEN 2 WHEN hs.timing_signal IN ('持有') THEN 3 WHEN hs.timing_signal IN ('弱势持有') THEN 4 ELSE 5 END, COALESCE(hs.rec_score, 0) DESC, COALESCE(hs.rr_ratio,0) DESC, hs.code """).fetchall() conn.close() stocks = [dict(r) for r in rows] return jsonify({"stocks": stocks, "total": len(stocks)}) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/tracking") def get_tracking(): """策略追踪评估：所有推荐的历史记录和结果""" import sqlite3 conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT st.*, lp.price as current_price FROM strategy_tracking st LEFT JOIN live_prices lp ON st.code = lp.code ORDER BY CASE st.status WHEN 'active' THEN 0 WHEN 'hit_tp' THEN 1 WHEN 'hit_sl' THEN 2 ELSE 3 END, st.tracked_at DESC """).fetchall() tracks = [] for r in rows: d = dict(r) # 理论盈亏：按买入区中值买入，当前价 vs 中值 if d.get('entry_mid') and d.get('current_price') and d.get('status') == 'active': mid = float(d['entry_mid']) price = float(d['current_price']) d['theoretical_pnl'] = round((price - mid) / mid * 100, 2) # 浮盈：有实操且 active 状态 if d.get('actual_entry') and d.get('actual_shares') and d.get('status') == 'active' and d.get('current_price'): entry = float(d['actual_entry']) price = float(d['current_price']) shares = int(d['actual_shares']) d['floating_pnl'] = round((price - entry) / entry * 100, 2) d['floating_amount'] = round((price - entry) * shares, 2) tracks.append(d) conn.close() return jsonify({ "tracks": tracks, "stats": { "total": len(tracks), "active": sum(1 for r in tracks if r["status"] == "active"), "hit_tp": sum(1 for r in tracks if r["status"] == "hit_tp"), "hit_sl": sum(1 for r in tracks if r["status"] == "hit_sl"), "expired": sum(1 for r in tracks if r["status"] == "expired"), "manual": sum(1 for r in tracks if r["status"] == "manual_close"), } }) @app.route("/api/research/strategies") def api_research_strategies(): """策略版本列表（含回测结果摘要，支持 period_tag 区间过滤）""" try: from strategy_lab import list_strategies, STRATEGY_DESCRIPTIONS pt = request.args.get('period_tag') strats = list_strategies(period_tag=pt) for s in strats: s['description'] = STRATEGY_DESCRIPTIONS.get(s['version'], {}) return jsonify({'strategies': strats}) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/api/research/backtest") def api_research_backtest(): """运行指定策略版本的回测""" from datetime import datetime, timedelta version = request.args.get('strategy', 'v4.1') period = request.args.get('period', '6m') market = request.args.get('market', 'all') # all | a | hk capital = float(request.args.get('capital', 1000000)) end_date = '2026-07-24' # 数据完整截止日 days = {'1m': 30, '6m': 185, '1y': 365, '2y': 730}.get(period, 185) start_date = (datetime.strptime(end_date, '%Y-%m-%d') - timedelta(days=days)).strftime('%Y-%m-%d') try: from strategy_lab import run_backtest, analyze_trade_list, save_analysis result = run_backtest(version, start_date, end_date, capital, universe=market) analysis = analyze_trade_list(result.get('trades', []), version) save_analysis(version, analysis) result['insights'] = analysis.get('insights', []) return jsonify(result) except ValueError as e: return jsonify({'error': str(e)}), 404 except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/api/research/analysis") def api_research_analysis(): """指定策略版本的因子归因分析""" version = request.args.get('strategy', 'v4.1') try: from strategy_lab import analyze_trades return jsonify(analyze_trades(version)) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/api/research/trades") def api_research_trades(): """指定策略版本的交易明细""" import sqlite3 as _sq, json as _json version = request.args.get('strategy', 'v4.1') conn = _sq.connect("/home/hmo/MoFin/data/mofin.db") row = conn.execute( "SELECT results_json FROM strategy_research WHERE version=? ORDER BY id DESC LIMIT 1", (version,)).fetchone() conn.close() if not row: return jsonify({'error': f'无 {version} 回测结果'}), 404 res = _json.loads(row[0]) return jsonify({'strategy': version, 'trades': res.get('trades', [])}) @app.route("/api/overview") def api_overview(): """概览数据""" try: from mofin_db import get_conn, query_holdings, query_portfolio_summary, query_latest_market conn = get_conn() holdings = query_holdings(conn) summary = query_portfolio_summary(conn) market = query_latest_market(conn) conn.close() if holdings: total_assets = summary.get("total_assets", 0) or 0 stock_value = summary.get("stock_value", 0) or 0 cash = summary.get("cash", 0) or 0 position_pct = summary.get("position_pct", 0) or 0 total_pnl = summary.get("total_pnl", 0) or 0 top_movers = sorted( [h for h in holdings if abs(h.get("change_pct", 0) or 0) >= 3], key=lambda x: abs(x.get("change_pct", 0) or 0), reverse=True)[:5] return jsonify({ "total_assets": total_assets, "stock_value": stock_value, "cash": cash, "position_pct": position_pct, "total_pnl": total_pnl, "top_movers": top_movers, "market": market, "alerts": _load_json(DATA_DIR / "alerts.json", [])[:10], "updated_at": summary.get("updated_at", ""), }) except Exception: return jsonify({"error": "数据库查询失败"}), 500 @app.route("/api/report/") def api_report(report_id): """单个报告详情""" # Try exact file first path = DATA_DIR / "reports" / f"{report_id}.json" if path.exists(): return jsonify(_load_json(path)) # Try prefix match reports_dir = DATA_DIR / "reports" if reports_dir.exists(): for f in reports_dir.iterdir(): if f.stem.startswith(report_id) and f.suffix == ".json": return jsonify(_load_json(f)) return jsonify({"error": "report not found"}), 404 @app.route("/api/stock/") def api_stock(code): """个股详情：DB策略(12维分析/action/三值RR)为主，JSON历史为辅。 根治：弹窗只读 data/stocks/{code}.json，与DB脱节，打开=空白。""" stock_data = _load_json(DATA_DIR / "stocks" / f"{code}.json", {}) try: conn = get_conn() conn.row_factory = sqlite3.Row r = conn.execute(""" SELECT hs.code, hs.name, hs.timing_signal, hs.action, hs.position_advice, hs.entry_low, hs.entry_high, hs.stop_loss, hs.take_profit, hs.rr_ratio, hs.rr_low, hs.rr_high, hs.full_analysis, hs.reassessed_at, hs.tag, hs.decision_type, hs.rec_score, lp.price AS live_price, lp.change_pct FROM holding_strategies hs LEFT JOIN live_prices lp ON hs.code = lp.code WHERE hs.code=? AND hs.status='active' """, (code,)).fetchone() conn.close() if r: stock_data.update({ "code": r["code"], "name": r["name"], "timing_signal": r["timing_signal"], "action": r["action"], "position_advice": r["position_advice"], "entry_low": r["entry_low"], "entry_high": r["entry_high"], "stop_loss": r["stop_loss"], "take_profit": r["take_profit"], "rr_ratio": r["rr_ratio"], "rr_low": r["rr_low"], "rr_high": r["rr_high"], "full_analysis": r["full_analysis"], "reassessed_at": r["reassessed_at"], "tag": r["tag"], "decision_type": r["decision_type"], "price": r["live_price"], "change_pct": r["change_pct"], }) except Exception as _e: stock_data.setdefault("db_error", str(_e)) return jsonify(stock_data) @app.route("/api/market") def api_market(): """市场观察""" try: from mofin_db import get_conn, query_latest_market conn = get_conn() data = query_latest_market(conn) conn.close() if data and data.get("sectors"): return jsonify(data) except Exception: pass return jsonify(_load_json(DATA_DIR / "market.json", {})) # ── 信号API（新增） ───────────────────────────────────── @app.route("/api/signals") def api_signals(): """最近信号 + 小果分析""" try: from mofin_db import get_conn conn = get_conn() signals = conn.execute(""" SELECT sn.id, sn.sector, sn.overall_sentiment, sn.summary, sn.source, sn.created_at, ss.signal_type, ss.severity FROM signal_news sn LEFT JOIN sector_signals ss ON sn.signal_id = ss.id ORDER BY sn.id DESC LIMIT 20 """).fetchall() conn.close() return jsonify([dict(r) for r in signals]) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/xiaoguo-scan") def api_xiaoguo_scan(): """小果扫描统计""" try: from mofin_db import get_conn conn = get_conn() total = conn.execute("SELECT COUNT(*) FROM xiaoguo_scan_tracker").fetchone()[0] found = conn.execute("SELECT COUNT(*) FROM xiaoguo_scan_tracker WHERE found_count>0").fetchone()[0] recent = conn.execute(""" SELECT code, name, last_scanned_at, found_count FROM xiaoguo_scan_tracker ORDER BY last_scanned_at DESC LIMIT 20 """).fetchall() source_count = conn.execute(""" SELECT source, COUNT(*) as cnt FROM signal_news WHERE datetime(created_at) > datetime('now', '-1 day') GROUP BY source """).fetchall() conn.close() return jsonify({ "total_scanned": total, "found_signals": found, "recent": [dict(r) for r in recent], "source_today": {r["source"]: r["cnt"] for r in source_count} }) except Exception as e: return jsonify({"error": str(e)}), 500 # ── 数据写入API ── @app.route("/api/update/portfolio", methods=["POST"]) def update_portfolio(): data = request.get_json(force=True) _save_portfolio(data) return jsonify({"status": "ok"}) @app.route("/api/update/watchlist", methods=["POST"]) def update_watchlist(): data = request.get_json(force=True) _save_watchlist(data) return jsonify({"status": "ok"}) @app.route("/api/update/report", methods=["POST"]) def update_report(): data = request.get_json(force=True) report_id = data.pop("_id", datetime.now().strftime("%Y%m%d_%H%M%S")) data["created_at"] = data.get("created_at", datetime.now().isoformat()) _save_json(DATA_DIR / "reports" / f"{report_id}.json", data) return jsonify({"status": "ok", "id": report_id}) @app.route("/api/update/stock/", methods=["POST"]) def update_stock(code): data = request.get_json(force=True) existing = _load_json(DATA_DIR / "stocks" / f"{code}.json", {}) history = existing.get("history", []) if data.get("entry"): history.append({ "time": datetime.now().isoformat(), "price": data.get("price"), "recommendation": data.get("recommendation"), "stop_loss": data.get("stop_loss"), "take_profit": data.get("take_profit"), "reason": data.get("reason"), }) existing.update(data) existing["history"] = history[-50:] _save_json(DATA_DIR / "stocks" / f"{code}.json", existing) return jsonify({"status": "ok"}) @app.route("/api/update/market", methods=["POST"]) def update_market(): data = request.get_json(force=True) or {} _save_json(DATA_DIR / "market.json", data) return jsonify({"status": "ok"}) # ── 知微分析结果写入API ── @app.route("/api/analysis/batch", methods=["POST"]) def analysis_batch(): """接收知微cron的分析结果，写回持仓/自选JSON的analysis字段""" data = request.get_json(force=True) or {} # 更新持仓 if "holdings" in data: pf = read_portfolio() idx = {h["code"]: i for i, h in enumerate(pf.get("holdings", []))} for item in data["holdings"]: code = item.get("code", "") if code not in idx: continue h = pf["holdings"][idx[code]] h["analysis"] = { "suggestion": item.get("suggestion"), "stop_loss": item.get("stop_loss"), "take_profit": item.get("take_profit"), "buy_zone_low": item.get("buy_zone_low"), "buy_zone_high": item.get("buy_zone_high"), "position_suggested": item.get("position_suggested"), "reason": item.get("reason"), "updated_at": datetime.now().isoformat(), } _save_portfolio(pf) # 更新自选 if "watchlist" in data: wl = read_watchlist() idx = {s["code"]: i for i, s in enumerate(wl.get("stocks", []))} for item in data["watchlist"]: code = item.get("code", "") if code not in idx: continue s = wl["stocks"][idx[code]] s["analysis"] = { "buy_low": item.get("buy_low"), "buy_high": item.get("buy_high"), "position_recommend": item.get("position_recommend"), "reason": item.get("reason"), "updated_at": datetime.now().isoformat(), } _save_watchlist(wl) return jsonify({"status": "ok", "updated_at": datetime.now().isoformat()}) # ── 操作决策库API ── @app.route("/api/decisions", methods=["GET"]) def get_decisions(): """返回决策库数据，统一新旧格式""" raw = read_decisions() decisions = raw.get("decisions", []) if not decisions and isinstance(raw, list): decisions = raw # portfolio 用来判断是持仓还是自选 portfolio = read_portfolio() watchlist = read_watchlist() holding_codes = {h.get("code","") for h in portfolio.get("holdings",[])} watch_codes = {s.get("code","") for s in watchlist.get("stocks",[])} normalized = [] for d in decisions: if not isinstance(d, dict): continue # 检测新旧格式：新格式有 stop_loss 顶层字段，旧格式有 trigger 对象 is_new = "stop_loss" in d and "trigger" not in d if is_new: code = d.get("code", "") name = d.get("name", "") price = d.get("price", 0) sl = d.get("stop_loss") tp = d.get("take_profit") el = d.get("entry_low") eh = d.get("entry_high") ts = d.get("tech_snapshot", "") # type: 持仓还是自选 if code in holding_codes: dtype = "持仓策略" elif code in watch_codes: dtype = "自选策略" else: dtype = "—" # 判断 active status_raw = d.get("status", "") status = "active" if status_raw in ("active", "updated", "") else "superseded" # trigger 对象 entry_zone_str = "" if el and eh: entry_zone_str = f"¥{el}~¥{eh}" elif el: entry_zone_str = f"≥¥{el}" trigger = {} if sl: trigger["stop_loss"] = f"¥{sl}" if isinstance(sl, (int,float)) else str(sl) if tp: trigger["take_profit"] = f"¥{tp}" if isinstance(tp, (int,float)) else str(tp) if entry_zone_str: trigger["entry_zone"] = entry_zone_str # current current = "" if price: current = f"现价¥{price}" if code and not code.startswith(("0","1")) else f"¥{price}" # zone_breach zone_breach = d.get("zone_breach", "") # updated_reason note = d.get("note", "") timing = d.get("timing_signal", "") reason_parts = [] if note: reason_parts.append(note) if timing and timing != "neutral": reason_parts.append(f"时机:{timing}") if d.get("rr_ratio"): reason_parts.append(f"盈亏比:{d['rr_ratio']}") # advice_timeline - 从新格式重建 timeline = [] entry = { "code": code, "name": name, "type": dtype, "status": status, "tag": d.get("tag", ""), "action": d.get("action", ""), "trigger": trigger, "current": current, "zone_breach": zone_breach, "updated_reason": " | ".join(reason_parts) if reason_parts else "", "advice_timeline": timeline, "changelog": d.get("changelog", []), "execution": d.get("execution", {}), "analysis": d.get("analysis", {}), "tech_snapshot": ts, "timestamp": d.get("timestamp", ""), "updated_by": "知微", } # 保留原始数据供前端扩展 entry["_raw_action"] = d.get("action", "") normalized.append(entry) else: # 旧格式：已有 trigger 等字段，直接保留 entry = dict(d) # 确保 status 正确 if entry.get("status") not in ("active", "superseded"): entry["status"] = "active" if not entry.get("type"): code = entry.get("code", "") if code in holding_codes: entry["type"] = "持仓策略" elif code in watch_codes: entry["type"] = "自选策略" else: entry["type"] = "—" normalized.append(entry) # 添加 execution 和 analysis 信息，按执行状态排序 for n in normalized: code = n.get("code", "") # 从原始数据中找到 execution 和 analysis raw_entry = next((d for d in decisions if isinstance(d, dict) and d.get("code") == code), {}) n["execution"] = raw_entry.get("execution", {"status": "none"}) n["analysis"] = raw_entry.get("analysis", {}) # 排序规则：推荐>执行中>观察>无标签 def sort_key(x): tag = x.get("tag", "") exec_status = x.get("execution", {}).get("status", "none") # 标签优先级（current_recommend才靠前，active_manual只是记录不升序） tag_order = {"current_recommend": 0} tag_priority = tag_order.get(tag, 50) # 执行状态优先级 exec_order = {"partial_exit": 0, "executing": 1, "observing": 2, "none": 99} exec_priority = exec_order.get(exec_status, 99) # 组合：先按标签排，再按执行状态排 return (tag_priority, exec_priority, x.get("code", "")) normalized.sort(key=sort_key) return jsonify({ "decisions": normalized, "total": len(normalized), "regenerated_at": raw.get("regenerated_at", ""), }) @app.route("/api/decisions/add", methods=["POST"]) def add_decision(): """新增/更新一条决策（新格式）""" data = request.get_json(force=True) or {} code = data.get("code", "") if not code: return jsonify({"status": "error", "message": "code required"}), 400 d = read_decisions() # 同一股票旧决策标记为superseded for e in d["decisions"]: if e["code"] == code and e.get("status") in ("active", "updated"): e["status"] = "superseded" entry = { "code": code, "name": data.get("name", ""), "price": data.get("price", 0), "action": data.get("action", ""), "stop_loss": data.get("stop_loss"), "take_profit": data.get("take_profit"), "entry_low": data.get("entry_low"), "entry_high": data.get("entry_high"), "tech_snapshot": data.get("tech_snapshot", ""), "timing_signal": data.get("timing_signal", ""), "rr_ratio": data.get("rr_ratio"), "tag": data.get("tag", ""), "note": data.get("note", ""), "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"), "updated_reason": data.get("updated_reason", ""), "status": "updated", "changelog": data.get("changelog", []), "execution": data.get("execution", {"status": "none"}), "analysis": data.get("analysis", {}), } d["decisions"].append(entry) _save_decision(code, entry.get('name',''), entry) return jsonify({"status": "ok", "entry": entry}) @app.route("/api/decisions/tag", methods=["POST"]) def set_decision_tag(): """设置/清除某只股票的推荐标签""" data = request.get_json(force=True) or {} code = data.get("code", "") tag = data.get("tag", "") # 'current_recommend', 'active_manual', or '' to clear if not code: return jsonify({"status": "error", "message": "code required"}), 400 d = read_decisions() found = False for e in d.get("decisions", []): if e.get("code") == code: e["tag"] = tag e["tag_updated"] = datetime.now().isoformat() found = True break if not found: return jsonify({"status": "error", "message": f"stock {code} not found"}), 404 _save_decision(code, e.get('name',''), e) return jsonify({"status": "ok", "code": code, "tag": tag}) @app.route("/api/decisions/pending") def get_pending_decisions(): """返回所有有未确认建议的条目""" d = read_decisions() pending = [] for entry in d["decisions"]: timeline = entry.get("advice_timeline", []) unconfirmed = [a for a in timeline if a.get("status") in (None, "pending")] if unconfirmed: pending.append({ "code": entry["code"], "name": entry["name"], "current": entry.get("current", ""), "pending_advice": unconfirmed, }) return jsonify(pending) @app.route("/api/advice/record", methods=["POST"]) def record_advice(): """记录一条分析建议，自动去重（相同code+同天+同方向=跳过）""" data = request.get_json(force=True) or {} code = data.get("code", "") if not code: return jsonify({"status": "error", "message": "code required"}), 400 direction = data.get("direction", "持有") today = datetime.now().strftime("%Y-%m-%d") d = read_decisions() entry = None for e in d["decisions"]: if e["code"] == code and e["status"] in ("active", "updated"): entry = e break if not entry: return jsonify({"status": "error", "message": f"no active decision for {code}"}), 404 timeline = entry.setdefault("advice_timeline", []) # 去重：同一天+同方向+摘要前40字相似 → 跳过 summary_short = (data.get("summary", "") or "")[:40] for a in timeline: a_date = a.get("date", "")[:10] a_dir = a.get("direction", "") a_summary = (a.get("summary", "") or "")[:40] if a_date == today and a_dir == direction and a_summary == summary_short: return jsonify({"status": "skipped", "reason": "duplicate", "advice": a}) advice = { "date": datetime.now().strftime("%Y-%m-%d %H:%M"), "direction": direction, "price": data.get("price", ""), "summary": data.get("summary", ""), "status": "pending", } timeline.append(advice) _save_decision(code, entry.get('name',''), entry) return jsonify({"status": "ok", "advice": advice}) @app.route("/api/advice/confirm", methods=["POST"]) def confirm_advice(): """确认/忽略/标记已执行""" data = request.get_json(force=True) or {} code = data.get("code", "") idx = data.get("index", -1) action = data.get("action", "confirmed") # confirmed | ignored | executed result = data.get("result", "") d = read_decisions() for e in d["decisions"]: if e["code"] == code and e["status"] == "active": timeline = e.get("advice_timeline", []) if 0 <= idx < len(timeline): timeline[idx]["status"] = action if action == "executed": timeline[idx]["evaluated"] = True timeline[idx]["evaluated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M") if result: timeline[idx]["result"] = result _save_decision(code, e.get('name',''), e) return jsonify({"status": "ok"}) return jsonify({"status": "error", "message": "not found"}), 404 # ── 准确率统计API ── @app.route("/api/stats/accuracy") def get_accuracy_stats(): data = _load_json(DATA_DIR / "accuracy_stats.json", {}) return jsonify(data) # ── 策略评估API ── @app.route("/api/evaluation") def get_evaluation(): """返回所有策略的双维度评估结果""" # 主数据源：evaluation.json eval_data = _load_json(DATA_DIR / "evaluation.json", {}) strategies = eval_data.get("strategies", []) if strategies: return jsonify(strategies) # 备选：从 decisions.json 的 evaluation 字段读取（尚未反写时的兼容） decisions = read_decisions() evals = [] for d in decisions.get("decisions", []): e = d.get("evaluation", []) if e: evals.append({ "code": d["code"], "name": d["name"], "type": d.get("type", ""), "current": d.get("current", ""), "evaluations": e, }) return jsonify(evals) @app.route("/api/evaluation/trigger", methods=["POST"]) def trigger_evaluation(): """手动触发策略评估""" import subprocess try: r = subprocess.run( ["python3", str(DATA_DIR.parent / "strategy_evaluator.py")], capture_output=True, timeout=60, text=True, ) return jsonify({"status": "ok", "output": r.stdout, "error": r.stderr}) except Exception as e: return jsonify({"status": "error", "message": str(e)}), 500 # ── 策略反馈API ── @app.route("/api/candidates") def get_candidates(): """候选股管道数据""" import sqlite3 conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT code, name, reason, entry_range, stop_loss, target, score_2nd, score_3rd, score_4th, score_5th, score_final, pass_s2, pass_s3, pass_s4, pass_s5, pass_final, promoted, promoted_at, log, created_at FROM candidates WHERE dropped IS NULL OR dropped=0 ORDER BY COALESCE(score_final,0) DESC, created_at DESC LIMIT 50 """).fetchall() conn.close() return jsonify([dict(r) for r in rows]) @app.route("/api/feedback") def get_feedback(): data = _load_json(DATA_DIR / "strategy_feedback.json", {}) return jsonify(data) # ── 持仓截图上传与解析 ──────────────────────────────── @app.route("/upload") def upload_page(): return send_from_directory(app.static_folder, "upload.html") def _ocr_image(image_path): """优先用小果GLM-OCR-8bit识别，失败则降级到pytesseract""" import sys from PIL import Image, ImageEnhance, ImageFilter import pytesseract # 尝试小果OCR（GLM-OCR-8bit） try: sys.path.insert(0, os.path.join(os.path.dirname(__file__), "scripts")) from ocr_client import ocr_image as xg_ocr result = xg_ocr(image_path, "请识别这张图片中所有文字，包括股票名称、代码、价格、持股数、金额、百分比等。输出完整内容。") if result.get("success") and len(result.get("text", "")) > 20: return result["text"].strip() except Exception: pass # 降级到tesseract # 降级：Tesseract（预处理优化中文表格识别） img = Image.open(image_path) # 预处理：放大 + 锐化 + 二值化，提升小字识别率 w, h = img.size if w < 2000 or h < 2000: scale = max(2, 2000 // min(w, h)) img = img.resize((w * scale, h * scale), Image.LANCZOS) # 转灰度 img = img.convert("L") # 增强对比度 enhancer = ImageEnhance.Contrast(img) img = enhancer.enhance(2.0) # 锐化 img = img.filter(ImageFilter.SHARPEN) # 二值化（自适应阈值） threshold = 128 img = img.point(lambda x: 255 if x > threshold else 0) # OCR：chip_sim+eng，PSM 6（统一文本块） text = pytesseract.image_to_string( img, lang="chi_sim+eng", config="--psm 6 --oem 3", ) return text.strip() ANALYZE_PROMPT = """你是股票持仓数据分析助手。以下是用户上传的持仓/自选截图经过OCR提取的文字，请从中提取所有股票信息。 判断这是「持仓截图」还是「自选截图」： - 持仓截图：每支股票有"证券数量"（持股数）、成本价、盈亏 - 自选截图：只有股票列表和价格，没有持股数/成本 股票代码格式： - A股：6位数字（如 600519, 000858, 300750） - 港股：纯数字代码（如 0700, 3690, 1211），不带HK前缀 ⚠️ 重要：截图顶部通常有汇总数据，如总资产、股票市值、可用资金、当日盈亏等。 如果OCR文字中有这些汇总数字，请一并提取到JSON的summary字段中。 不要自己计算汇总值，直接从OCR原文中提取。 请严格按照以下JSON格式回复，只输出JSON： ```json { "type": "portfolio" 或 "watchlist", "summary": { "total_assets": "总资产数字（可选，从截图中提取）", "stock_value": "股票市值/持仓市值数字（可选，从截图中提取）", "cash": "可用资金/现金数字（可选，从截图中提取）", "day_pnl": "当日盈亏金额（可选，从截图中提取）" }, "stocks": [ { "code": "股票代码", "name": "股票名称（中文）", "price": "现价（数字）", "shares": "持股数量（数字，持仓截图才有）", "cost": "成本价（数字，持仓截图才有）", "pnl": "盈亏百分比如+15.1%（持仓截图才有）", "position_pct": "仓位占比数字如12.5（可选）" } ] } ``` OCR原文： """ @app.route("/api/upload/analyze", methods=["POST"]) def upload_analyze(): """接收图片，OCR提取文字 → LLM解析结构化数据""" if "image" not in request.files: return jsonify({"error": "请上传图片"}), 400 f = request.files["image"] if not f.filename: return jsonify({"error": "空文件"}), 400 # 保存到临时目录 UPLOAD_DIR.mkdir(parents=True, exist_ok=True) ext = Path(f.filename).suffix or ".png" save_path = UPLOAD_DIR / f"{uuid.uuid4().hex}{ext}" f.save(str(save_path)) try: # 第一步：OCR提取文字 raw_text = _ocr_image(str(save_path)) if not raw_text: return jsonify({"error": "OCR未识别到文字，请确认图片清晰"}), 400 except Exception as e: os.unlink(str(save_path)) return jsonify({"error": f"OCR失败: {e}"}), 500 # 第二步：LLM解析结构化数据（走文本API，不走视觉） llm_text = _llm_parse(raw_text, ANALYZE_PROMPT) os.unlink(str(save_path)) # 从LLM回复中提取JSON json_match = re.search(r"```(?:json)?\s*({.*?})\s*```", llm_text, re.DOTALL) if json_match: try: parsed = json.loads(json_match.group(1)) except json.JSONDecodeError: return jsonify({"error": f"LLM解析JSON失败: {llm_text[:500]}"}), 500 else: # 尝试直接找JSON（没被代码块包裹） try: parsed = json.loads(llm_text) except json.JSONDecodeError: return jsonify({"error": f"未提取到结构化数据: {raw_text[:300]}...\n\nLLM回复: {llm_text[:500]}"}), 500 return jsonify(parsed) def _llm_parse(text, prompt_template): """发送OCR文本到Hermes LLM解析，返回JSON字符串""" payload = json.dumps({ "model": "hermes-agent", "messages": [ {"role": "system", "content": "你是一个数据提取助手。从OCR文字中提取结构化JSON数据。"}, {"role": "user", "content": prompt_template + "\n" + text}, ], "max_tokens": 4096, }).encode() req = urllib.request.Request(GATEWAY, data=payload, method="POST") req.add_header("Content-Type", "application/json") req.add_header("Authorization", f"Bearer {API_KEY}") req.add_header("X-Hermes-Session-Id", "upload-ocr-parse") try: resp = urllib.request.urlopen(req, timeout=120) data = json.loads(resp.read()) return data.get("choices", [{}])[0].get("message", {}).get("content", "") except Exception as e: return f"ERROR: {e}" @app.route("/api/upload/confirm", methods=["POST"]) def upload_confirm(): """确认解析结果，更新数据文件""" data = request.get_json(force=True) stocks = data.get("stocks", []) doc_type = data.get("type", "portfolio") # 尝试获取实时行情补充数据 try: codes = [s["code"] for s in stocks if s.get("code")] if codes: # DB 优先（price_monitor 维护的实时价） db_prices = {} try: import sqlite3 db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db') db.row_factory = sqlite3.Row for code in codes: row = db.execute("SELECT price, change_pct FROM holdings WHERE code=? AND is_active=1", (code,)).fetchone() if row and row['price']: db_prices[code] = (row['price'], row['change_pct'] or 0) db.close() except Exception: pass # Fallback: 腾讯 API need_tencent = [c for c in codes if c not in db_prices] if need_tencent: qs = " ".join( f"hk{c}" if len(c) == 5 else f"sz{c}" if c.startswith("0") or c.startswith("3") else f"sh{c}" if c.startswith("6") else f"hk{c}" for c in need_tencent ) url = f"https://qt.gtimg.cn/q={qs}" req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) resp = urllib.request.urlopen(req, timeout=10) qt_text = resp.read().decode("gbk", errors="replace") # 优先 DB 价格，再补腾讯 for stock in stocks: code = stock.get("code", "") if code in db_prices: if not stock.get("price"): stock["price"] = db_prices[code][0] elif need_tencent and code in need_tencent: prefix = "hk" if len(code) == 5 else "sz" if code.startswith(("0","3")) else "sh" if code.startswith("6") else "hk" m = re.search(rf'{prefix}{code}="([^"]+)"', qt_text) if m: fields = m.group(1).split('~') if not stock.get("name"): stock["name"] = fields[1] if not stock.get("price"): stock["price"] = fields[3] except: pass # 行情获取失败不影响主流程 # 更新对应数据文件 if doc_type == "portfolio": existing = read_portfolio() old_holdings = {h["code"]: h for h in existing.get("holdings", []) if h.get("code")} new_holdings = [] for s in stocks: code = s.get("code", "") old = old_holdings.get(code, {}) new_shares = int(s["shares"]) if str(s.get("shares", "")).lstrip('-').isdigit() else old.get("shares", 0) old_shares = old.get("shares", 0) # 股数突变检测：旧200→新0是合理卖出，但旧0→新200可能是OCR错读 if old_shares > 0 and new_shares == 0 and old_shares != new_shares: print(f"[仓位变动] {code} {s.get('name','')}: {old_shares}→{new_shares} (卖出清仓)") elif abs(new_shares - old_shares) > max(old_shares * 0.5, 100) and old_shares > 0: print(f"[仓位变动] {code} {s.get('name','')}: {old_shares}→{new_shares} (变动较大)") new_holdings.append({ "code": code, "name": s.get("name") or old.get("name", ""), "shares": new_shares, "price": float(s.get("price", 0)) or old.get("price", 0), "cost": float(s.get("cost", 0)) if s.get("cost") else old.get("cost", 0), "pnl": s.get("pnl") or old.get("pnl", ""), "position_pct": float(s.get("position_pct", 0)) if s.get("position_pct") else old.get("position_pct", 0), "change_pct": old.get("change_pct", 0), }) existing["holdings"] = new_holdings # 使用截图中的汇总数据（优先），没有则用旧数据 summary = data.get("summary", {}) if summary.get("stock_value"): existing["stock_value"] = float(summary["stock_value"]) else: existing["stock_value"] = round( sum(h["shares"] * h["price"] for h in existing["holdings"]), 2 ) if summary.get("cash"): existing["cash"] = float(summary["cash"]) if summary.get("total_assets"): existing["total_assets"] = float(summary["total_assets"]) else: # Use unified formula (includes frozen_cash) from mo_models import calc_total_assets existing["total_assets"] = calc_total_assets(existing) if summary.get("day_pnl"): existing["day_pnl"] = float(summary["day_pnl"]) existing["updated_at"] = datetime.now().isoformat() # 计算仓位% if existing["total_assets"] > 0: existing["position_pct"] = round(existing["stock_value"] / existing["total_assets"] * 100, 2) _save_portfolio(existing) msg = f"更新了 {len(stocks)} 只持仓股" elif doc_type == "watchlist": existing = read_watchlist() existing["stocks"] = [ { "code": s.get("code", ""), "name": s.get("name", ""), "price": float(s.get("price", 0)) if s.get("price") else 0, } for s in stocks ] existing["updated_at"] = datetime.now().isoformat() _save_watchlist(existing) msg = f"更新了 {len(stocks)} 只自选股" else: return jsonify({"error": f"未知类型: {doc_type}"}), 400 return jsonify({"status": "ok", "message": msg}) # ── TDX中继实时行情接收API ── @app.route("/api/update/realtime", methods=["POST"]) def update_realtime(): """接收小小莫中继的实时行情数据""" data = request.get_json(force=True) or {} stocks = data.get("stocks", []) source = data.get("source", "unknown") if not stocks: return jsonify({"status": "error", "message": "没有股票数据"}), 400 # 更新 portfolio.json 中的实时价格（change_pct字段） pf = read_portfolio() pf_holdings = {h["code"]: h for h in pf.get("holdings", [])} updated = 0 for s in stocks: code = s.get("code", "") if code in pf_holdings: pf_holdings[code]["price"] = float(s.get("price", pf_holdings[code].get("price", 0))) pf_holdings[code]["change_pct"] = float(s.get("change_pct", 0)) pf_holdings[code]["high"] = float(s.get("high", 0)) pf_holdings[code]["low"] = float(s.get("low", 0)) pf_holdings[code]["open"] = float(s.get("open", 0)) pf_holdings[code]["volume"] = int(s.get("volume", 0)) pf_holdings[code]["data_source"] = source pf_holdings[code]["updated_at"] = datetime.now().isoformat() updated += 1 # 也更新 watchlist.json wl = read_watchlist() wl_stocks = {s["code"]: s for s in wl.get("stocks", [])} for s in stocks: code = s.get("code", "") if code in wl_stocks: wl_stocks[code]["price"] = float(s.get("price", wl_stocks[code].get("price", 0))) wl_stocks[code]["change_pct"] = float(s.get("change_pct", 0)) pf["updated_at"] = datetime.now().isoformat() wl["updated_at"] = datetime.now().isoformat() _save_portfolio(pf) _save_watchlist(wl) return jsonify({ "status": "ok", "updated": updated, "source": source, "timestamp": datetime.now().isoformat(), }) # ── Dashboard 管理门户 ────────────────────────────── @app.route("/dashboard") def dashboard_page(): return send_from_directory(str(Path(__file__).parent / "templates"), "dashboard.html") @app.route("/api/health") def api_health(): return jsonify({"status": "ok", "uptime": int(time.time() - START_TIME)}) @app.route("/api/services") def api_services(): result = [] for svc in DASH_SERVICES: ok = _check_svc(svc) result.append({ "name": svc["name"], "label": svc["label"], "port": svc["port"], "type": svc["type"], "layer": svc["layer"], "critical": svc["critical"], "health": {"ok": ok}, }) ok_count = sum(1 for s in result if s["health"]["ok"]) return jsonify({"services": result, "summary": {"ok": ok_count, "total": len(result)}}) @app.route("/api/expected") def api_expected(): expected = [{ "name": s["name"], "label": s["label"], "port": s["port"], "expected": "running", "critical": s["critical"], "layer": s["layer"], "check": f"{s['type']}:{s['port']}" if s["port"] else s["type"], } for s in DASH_SERVICES] actual = {} for svc in DASH_SERVICES: actual[svc["name"]] = "running" if _check_svc(svc) else "stopped" return jsonify({"expected": expected, "actual": actual}) @app.route("/api/monitor") def api_monitor(): tasks = [] tier1 = {"summary": {"ok": 0, "total": 0}, "services": []} tier2 = {"summary": {"ok": 0, "total": 0}, "services": []} t1_path = GATEWAY_TEMP / "last_health_check.json" if t1_path.exists(): try: with open(t1_path, encoding="utf-8") as f: tier1 = json.load(f) tasks.append({"name": "agents-health-check", "status": "cron_ok"}) except Exception: tasks.append({"name": "agents-health-check", "status": "error"}) else: tasks.append({"name": "agents-health-check", "status": "not_deployed"}) t2_path = GATEWAY_TEMP / "last_daily_health.json" if t2_path.exists(): try: with open(t2_path, encoding="utf-8") as f: tier2 = json.load(f) tasks.append({"name": "agents-daily-health", "status": "cron_ok"}) except Exception: tasks.append({"name": "agents-daily-health", "status": "error"}) else: tasks.append({"name": "agents-daily-health", "status": "not_deployed"}) tasks.append({"name": "dashboard", "status": "running"}) return jsonify({ "tasks": tasks, "tier1": tier1, "tier2": tier2, "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }) @app.route("/api/module-spec/") def api_module_spec(module): spec_path = SPECS_DIR / f"{module.replace('..', '').replace('/', '').replace(chr(92), '')}.json" if spec_path.exists(): try: with open(spec_path, encoding="utf-8") as f: return jsonify(json.load(f)) except Exception as e: return jsonify({"error": str(e)}), 500 return jsonify({"error": f"Module '{module}' not found"}), 404 # ── XMPP 通信监控 API ───────────────────────────────── @app.route("/api/xmpp/messages") def api_xmpp_messages(): """查询 XMPP 消息日志""" since = request.args.get("since", "") agent = request.args.get("agent", "") status = request.args.get("status", "") limit = int(request.args.get("limit", 50)) try: from xmpp_logger import query msgs = query(since=since or None, agent=agent or None, status=status or None, limit=limit) return jsonify({"messages": msgs, "total": len(msgs)}) except ImportError: return jsonify({"messages": [], "total": 0}) @app.route("/api/xmpp/health") def api_xmpp_health(): """XMPP 通道健康检查""" try: from xmpp_logger import health as xmpp_health h = xmpp_health() # 补充 ejabberd Docker 状态 import subprocess r = subprocess.run(["docker", "ps", "--filter", "name=ejabberd", "--format", "{{.Status}}"], capture_output=True, timeout=5, text=True) h["ejabberd"] = r.stdout.strip() if r.returncode == 0 and r.stdout.strip() else "not_found" return jsonify(h) except ImportError: return jsonify({"status": "no_logger", "last_message_age_sec": -1, "error_rate_1h": 0, "ejabberd": "unknown"}) @app.route("/api/xmpp/stats") def api_xmpp_stats(): """XMPP 消息统计""" try: from xmpp_logger import stats as xmpp_stats return jsonify(xmpp_stats()) except ImportError: return jsonify({"today": {"sent": 0, "failed": 0, "latency_avg": 0}, "week": {"sent": 0, "failed": 0}}) @app.route("/api/xmpp/autoheal", methods=["GET", "POST"]) def api_xmpp_autoheal(): """自愈：检测异常并自动修复。GET=查看状态, POST=执行修复""" try: from xmpp_logger import auto_heal if request.method == "POST": result = auto_heal() return jsonify(result) else: return jsonify({"usage": "POST to trigger auto-heal"}) except ImportError: return jsonify({"error": "xmpp_logger not available"}), 500 @app.route("/api/xmpp/keys") def api_xmpp_keys(): """API Key 可用性：从 AgentsMeeting 获取并选择最佳 key""" try: from xmpp_logger import best_key bk = best_key() return jsonify({"best_key": bk} if bk else {"error": "no keys available"}) except ImportError: return jsonify({"error": "xmpp_logger not available"}), 500 # ── 开发原则 Tab 端点（仿 AgentsMeeting: G规范/K测试/H需求）── DOCS_DIR = Path(__file__).resolve().parent / "docs" HEALTH_REPORT = Path(__file__).resolve().parent / "gateway" / "temp" / "last_health_check.json" @app.route("/api/spec") def api_spec(): """G 规范：docs/dev-spec.md 内容""" f = DOCS_DIR / "dev-spec.md" if not f.exists(): return jsonify({"ok": False, "error": "dev-spec.md not found"}) return jsonify({"ok": True, "content": f.read_text(encoding="utf-8")}) @app.route("/api/spec/history") def api_spec_history(): """dev-spec.md 的 git 历史""" import subprocess try: r = subprocess.run( ["git", "log", "--oneline", "-20", "--", "docs/dev-spec.md"], capture_output=True, timeout=10, text=True, cwd=str(Path(__file__).resolve().parent)) lines = [l for l in r.stdout.splitlines() if l.strip()] return jsonify({"ok": True, "log": lines, "count": len(lines)}) except Exception as e: return jsonify({"ok": False, "error": str(e)[:100]}) @app.route("/api/tests") def api_tests(): """K 测试：agents_health_check 服务检查结果，渲染为 pass/fail 测试报告""" if not HEALTH_REPORT.exists(): return jsonify({"ok": False, "error": "health report not found (cron not run yet)"}) try: rep = json.loads(HEALTH_REPORT.read_text(encoding="utf-8")) except Exception as e: return jsonify({"ok": False, "error": str(e)[:100]}) tests = [] for svc in rep.get("services", []): ok = svc.get("health", {}).get("ok", False) tests.append({ "name": f"{svc.get('label', svc.get('name'))} ({svc.get('name')})", "ok": ok, "expected": False, "detail": svc.get("detail", ""), }) passed = sum(1 for t in tests if t["ok"]) return jsonify({ "ok": True, "tests": tests, "summary": {"total": len(tests), "passed": passed, "failed": len(tests) - passed}, "time": rep.get("generated_at", ""), }) @app.route("/api/prd") def api_prd(): """H 需求：docs/prd.md（不存在则返回未建立）""" f = DOCS_DIR / "prd.md" if not f.exists(): return jsonify({"ok": False, "error": "prd.md 尚未建立"}) return jsonify({"ok": True, "content": f.read_text(encoding="utf-8")}) # 注册提示词管理路由 register_routes(app) @app.route("/api/evolution/dashboard") def api_evolution_dashboard(): """进化模块 Dashboard""" try: sys.path.insert(0, '/home/hmo/MoFin/evolution') from evolution_api import get_evolution_dashboard return jsonify(get_evolution_dashboard()) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route("/api/evolution/health_trend") def api_evolution_health_trend(): """健康度趋势""" try: version = request.args.get('version', 'v_next4') days = int(request.args.get('days', 30)) sys.path.insert(0, '/home/hmo/MoFin/evolution') from evolution_api import get_health_trend return jsonify({'trend': get_health_trend(version, days)}) except Exception as e: return jsonify({'error': str(e)}), 500 if __name__ == "__main__": port = int(os.environ.get("PORT", 8899)) print(f"🚀 MoFin Dashboard → http://0.0.0.0:{port}") app.run(host="0.0.0.0", port=port, debug=False)