diff --git a/archive/20260722-scripts-cleanup/MANIFEST.json b/archive/20260722-scripts-cleanup/MANIFEST.json new file mode 100644 index 00000000..a871cbed --- /dev/null +++ b/archive/20260722-scripts-cleanup/MANIFEST.json @@ -0,0 +1,188 @@ +{ + "A_shadow_duplicates": [ + "300308_monitor.py", + "_watchdog_report.py", + "accumulation_scanner.py", + "batch_reassess.py", + "branch_evaluator.py", + "candidate_filter.py", + "capital_flow_collector.py", + "check-prompt-deps.py", + "chip_factors.py", + "clean_watchlist.py", + "cron_health_monitor.py", + "data_flow_audit.py", + "data_governance.py", + "divergence_detector.py", + "fix_gateway_port.py", + "generate_report.py", + "get_realtime_prices.py", + "hardcode_scanner.py", + "hk_rate.py", + "holdings_reconciliation.py", + "import_full_stocks.py", + "import_holding_xls.py", + "intraday_health_check.py", + "macro_context_collector.py", + "macro_signal_consumer.py", + "market_scanner.py", + "market_screener.py", + "market_watch.py", + "meta_growth.py", + "mo_alphasift_bridge.py", + "mo_bridge.py", + "mo_config.py", + "mo_data.py", + "mo_dsa_opinion.py", + "mo_models.py", + "mo_provider.py", + "mofin_collect.py", + "mofin_db.py", + "mofin_news.py", + "mofin_query.py", + "monitor_300308.py", + "morning_health_check.py", + "multi_timeframe.py", + "ocr_client.py", + "per_stock_reassess.py", + "pre-flight-check.py", + "preflight_verify.py", + "price_monitor.py", + "refresh_macro_context.py", + "refresh_mtf_cache.py", + "review_needed_watchdog.py", + "run_all_tests.py", + "self_todo_executor.py", + "session_to_cron_bridge.py", + "stale_detector.py", + "stale_push_wlin.py", + "stock_profile.py", + "stock_quote.py", + "stock_scorer.py", + "stock_sector_enrich.py", + "strategy-staleness-check.py", + "strategy_feedback.py", + "strategy_lifecycle.py", + "strategy_review.py", + "strategy_summary.py", + "strategy_tree.py", + "sync_cron_prompts.py", + "sync_dashboard.py", + "sync_decisions_to_db.py", + "technical_analysis.py", + "trend_detector.py", + "vacuum_state_db.py", + "verify_reassess_pipeline.py", + "watchlist_auto_exit.py" + ], + "B_unref_tools": [ + "add_hygiene_cron.py", + "analyze_health.py", + "archive_legacy_json.py", + "archive_old_files.py", + "audit_data.py", + "audit_deadcode.py", + "audit_duplication.py", + "audit_runtime.py", + "backfill_price_events.py", + "check_3_dbs.py", + "check_97_bug.py", + "check_backfill_job.py", + "check_candidates_schema.py", + "check_current_errors.py", + "check_dashboard.py", + "check_db_paths.py", + "check_db_state.py", + "check_default_pool.py", + "check_error_freshness.py", + "check_fk.py", + "check_llm_jobs.py", + "check_missing_scripts.py", + "check_new_imports.py", + "check_price_events.py", + "check_srv.py", + "check_stocks_table.py", + "check_table_schemas.py", + "check_todos_schema.py", + "check_triggered.py", + "classify_diverged.py", + "cleanup_disabled_cron.py", + "cron_status.py", + "db_recovery.py", + "dump_health.py", + "find_97_rows.py", + "find_cron_errors.py", + "fix_crontab_cron_to_xmpp.py", + "fix_crontab_market.py", + "fix_self_todo_path.py", + "fix_symlinks.py", + "fix_todos_db.py", + "full_error_inventory.py", + "get_full_errors.py", + "get_today_errors.py", + "inspect_third_db.py", + "inv5.py", + "investigate_5_issues.py", + "kanban_create.py", + "kanban_fix_status.py", + "kanban_inspect.py", + "key_status.py", + "list_tables.py", + "merge_third_db.py", + "notify_user.py", + "notify_user2.py", + "notify_user3.py", + "notify_user4.py", + "notify_user5.py", + "notify_user6.py", + "notify_user7.py", + "pp.py", + "register_l134_jobs.py", + "remove_cron_watchdog.py", + "remove_xiaoguo.py", + "remove_xiaoguo2.py", + "sc2.py", + "sc3.py", + "sc5.py", + "syntax_check.py", + "test_auto_heal.py", + "test_bot_import.py", + "test_db_only.py", + "test_db_write.py", + "test_dual_write.py", + "test_freshness_query.py", + "test_image_pipeline.py", + "test_key7_guard.py", + "test_llm.py", + "test_msg_log.py", + "test_multi_profile.py", + "test_noping.py", + "test_ocr_pipeline_live.py", + "test_production.py", + "test_raw_insert.py", + "test_rerun_action.py", + "test_single.py", + "test_sn_ocr2.py", + "test_xmpp_logger.py", + "test_zone_gate.py", + "trace_97_source.py", + "trigger_weekend_jobs.py", + "unify_sources.py", + "update_backfill_cron.py", + "validate_fixes.py", + "verify_300308.py", + "verify_deployment.py", + "verify_health_fix.py", + "verify_health_json.py", + "verify_self_check_section.py", + "verify_self_todo.py", + "xiaoguo_news_processor.py", + "xiaoguo_scanner.py", + "xiaoguo_sentiment_bridge.py", + "xiaoguo_signal_consumer.py" + ], + "keep": [ + "prepare_report_data.py", + "server.py" + ] +} \ No newline at end of file diff --git a/archive/20260722-scripts-cleanup/batch_reassess.py b/archive/20260722-scripts-cleanup/batch_reassess.py index 196dab1b..450eee54 100644 --- a/archive/20260722-scripts-cleanup/batch_reassess.py +++ b/archive/20260722-scripts-cleanup/batch_reassess.py @@ -1,19 +1,30 @@ #!/usr/bin/env python3 -"""batch_reassess.py — 批量补全九维分析（逐只处理，间隔防限流） +"""batch_reassess.py — 批量补全12维(九维矩阵)LLM分析（逐只处理，间隔防限流） -用法: python3 batch_reassess.py [--all] [--code XXXXXX] +用法: + python3 batch_reassess.py # 所有缺分析/过期的 active 策略 + python3 batch_reassess.py --type holding # 只处理持仓策略 + python3 batch_reassess.py --type watchlist # 只处理自选策略 + python3 batch_reassess.py --type holding --today # 持仓每日刷新（今早未评过的强制重评） + python3 batch_reassess.py --code XXXXXX # 单只 -流程：收集最新数据 → 调LLM(gateway)写九维分析+策略 → 保存到DB +流程：收集最新数据 → 调LLM(gateway)写12维分析+策略 → 保存到DB """ -import sys, json, subprocess, sqlite3, re, time +import sys, json, subprocess, sqlite3, re, time, os from datetime import datetime +# ── 共享 LLM 客户端 + DB 工具（profile-scripts 硬链到同目录）── +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, "/home/hmo/MoFin") +from llm_client import call_llm, REASSESS_MODEL, FALLBACK_MODEL, gateway_alive, ocg_alive +from mofin_db import snapshot_strategy_history, sync_recommend_tag + DB = "/home/hmo/MoFin/data/mofin.db" -GATEWAY = "http://127.0.0.1:8643/v1/chat/completions" COOLDOWN_HOURS = 1 +STALE_HOURS = 20 # 分析超过20小时视为过期，需要重评 def has_llm_analysis(code): - """检查是否为LLM生成的九维分析（>500字）""" + """检查是否为LLM生成的12维分析（>500字）""" conn = sqlite3.connect(DB) r = conn.execute("SELECT LENGTH(full_analysis) FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() conn.close() @@ -33,13 +44,41 @@ def in_cooldown(code): except: return False +def analysis_stale(code, force_today=False): + """分析是否过期（>STALE_HOURS 或 force_today 时今早4点前未重评）""" + conn = sqlite3.connect(DB) + r = conn.execute("SELECT reassessed_at FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() + conn.close() + if not r or not r[0]: + return True + try: + last = datetime.fromisoformat(r[0]) + if force_today: + today4am = datetime.now().replace(hour=4, minute=0, second=0, microsecond=0) + return last < today4am + return (datetime.now() - last).total_seconds() / 3600 > STALE_HOURS + except: + return True + +def get_portfolio(): + """从 portfolio_summary 读实时现金/总资产（不再硬编码）""" + try: + conn = sqlite3.connect(DB) + r = conn.execute("SELECT cash, total_assets FROM portfolio_summary WHERE id=1").fetchone() + conn.close() + if r and r[1]: + return int(r[0] or 0), int(r[1]) + except Exception: + pass + return 0, 0 + def collect_data(code): - """收集最新数据""" + """收集最新数据（含完整策略原文）""" data = {"code": code} - # 从DB读策略 + # 从DB读策略（含 full_analysis / changelog_json / position_advice） conn = sqlite3.connect(DB) - r = conn.execute("SELECT name, entry_low, entry_high, stop_loss, take_profit, timing_signal, action, rr_ratio, tech_snapshot, sector_context, stock_category FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() + r = conn.execute("SELECT name, entry_low, entry_high, stop_loss, take_profit, timing_signal, action, rr_ratio, tech_snapshot, sector_context, stock_category, full_analysis, changelog_json, reassessed_at, position_advice FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() if r: data["name"] = r[0] data["entry_low"] = r[1] or 0 @@ -52,10 +91,21 @@ def collect_data(code): data["tech_snapshot"] = r[8] or "" data["sector_context"] = r[9] or "" data["stock_category"] = r[10] or "" + data["full_analysis"] = r[11] or "" + data["changelog_json"] = r[12] or "" + data["reassessed_at"] = r[13] or "" + data["position_advice"] = r[14] or "" conn.close() # 从腾讯API拉最新价和基本面 - prefix = "sh" if str(code).startswith(("6","9")) else "sz" + # 代码前缀：5位=港股(hk)，6/9开头=沪(sh)，其他=深(sz) + _c = str(code) + if len(_c) == 5: + prefix = "hk" + elif _c.startswith(("6", "9")): + prefix = "sh" + else: + prefix = "sz" try: r = subprocess.run(["curl", "-s", f"http://qt.gtimg.cn/q={prefix}{code}"], capture_output=True, timeout=10) parts = r.stdout.decode("gbk", errors="ignore").split("~") @@ -80,9 +130,10 @@ def collect_data(code): return data def build_prompt(data): - """构建LLM prompt，要求输出完整策略""" - cash = 321271 # 可用现金（从DB读取） - total = 952879 # 总资产 + """构建LLM prompt，先审阅原策略再结合实时数据输出修改判断+九维矩阵分析""" + cash, total = get_portfolio() + if not total: + cash, total = 241330, 929727 # 兜底（DB读不到时） # 拉取资金流数据 _flow_note = "暂无资金流数据" @@ -122,7 +173,53 @@ def build_prompt(data): except: pass - return f"""你是一个资深A股分析师。请对{data['code']} {data.get('name','')}做一个完整的九维矩阵分析，并输出策略参数。 + # ── 构建【原策略全文】section ── + _params_parts = [] + if data.get('action'): _params_parts.append(f"当前策略: {data['action']}") + if data.get('timing_signal'): _params_parts.append(f"信号: {data['timing_signal']}") + if data.get('entry_low') or data.get('entry_high'): + _params_parts.append(f"买入区间: {data.get('entry_low',0)}~{data.get('entry_high',0)}") + if data.get('stop_loss'): _params_parts.append(f"止损: {data['stop_loss']}") + if data.get('take_profit'): _params_parts.append(f"止盈: {data['take_profit']}") + if data.get('position_advice'): _params_parts.append(f"仓位: {data['position_advice']}") + _params_str = " | ".join(_params_parts) if _params_parts else "无策略参数" + + # 最近3条变更记录 + _changelog_str = "无变更记录" + try: + _cl_raw = data.get('changelog_json', '') + if _cl_raw: + _cl = json.loads(_cl_raw) if isinstance(_cl_raw, str) else _cl_raw + if isinstance(_cl, list) and _cl: + _recent = _cl[-3:] if len(_cl) > 3 else _cl + _cl_lines = [] + for i, c in enumerate(_recent): + _act = c.get('action', c.get('reason', '')) if isinstance(c, dict) else str(c) + _ts = c.get('timestamp', '') if isinstance(c, dict) else '' + _cl_lines.append(f" {i+1}. {_ts[:16]} {_act[:80]}") + if _cl_lines: + _changelog_str = "\n".join(_cl_lines) + except: + pass + + # 完整分析原文（不截断） + _full_analysis = data.get('full_analysis', '') or '' + _fa_display = _full_analysis if _full_analysis else '（首次分析，无历史）' + + _orig_strategy_section = f"""当前策略参数: {_params_str} + +变更记录（最近3条）: +{_changelog_str} + +完整分析原文: +{_fa_display}""" + + return f"""你是一个资深A股分析师。请先审阅以下【原策略全文】，判断是否需要修改策略，然后做出完整的九维矩阵分析。 + +【原策略全文】 +{_orig_strategy_section} + +── 以上是已有的策略，以下是当前实时数据，请结合两者做出判断 ── ⚠️ 重要：以下9个维度不是独立分析的，你必须交叉对比后给出综合结论。 例如：如果消息面利好但资金流在流出，说明利好可能是出货；如果基本面强但技术面破位，说明估值可能还没到底。 @@ -136,11 +233,18 @@ PE={data.get('pe','?')}（最新财报） 市值={data.get('mcap','?')}亿 资金流：{_flow_note}（近5日累计） 消息面：{_news_note}（最近3条，自动标注抓取时间） 当前信号：{data.get('timing_signal','?')} 分类：{data.get('stock_category','?')} -原策略：{(data.get('action','') or '')[:200]} 我的总资产={total}元，可用现金={cash}元。 -请严格按以下格式输出： +请严格按以下格式输出（注意节标题不可省略）： + +【维持或修改】明确二选一判断：维持原策略 / 需要修改策略 +【修改点及理由】 +如果维持原策略 → 写"无需修改" +如果需要修改 → 逐条列出（每条格式："- 修改点名称：理由说明"） +【最终新策略】 +用自然语言输出完整的最终策略全文（200-400字），自包含核心交易逻辑、买入区间价格、止损价、止盈价、仓位比例、风险提示。 +⚠️ 本段不要使用【综合结论】【买入区间】等标签——用自然语言描述即可。 【交叉分析】用2-3句话说明哪些维度出现矛盾/共振，最关键的信号是什么 ① 大盘×基本面 [一句话，说明矛盾关系] @@ -162,13 +266,18 @@ PE={data.get('pe','?')}（最新财报） 市值={data.get('mcap','?')}亿 【建议止损】数字 【建议止盈】数字 -【建议仓位】只有综合结论为"买入"时才输出此项。仓位计算公式： +【建议仓位】⚠️不可省略。综合结论非"买入"时写"不新建仓"；为"买入"时按以下公式： 基础仓位按RR确定：RR<1.5→不推荐，RR1.5~3→8%，RR3~5→12%，RR5+→15% 大盘偏弱×0.8，大盘偏强×1.15 蓝筹/白马×1.2，成长×0.85，题材/短线×0.6 最终仓位范围：5%~20% 同时考虑：现金{cash}元足够买多少手。 -输出格式："X%（理由：一句话说明为什么这个仓位）""" +输出格式："X%（理由：一句话说明为什么这个仓位）" + +⚠️ 输出纪律（必须遵守）： +1. 直接以【维持或修改】开头，禁止任何寒暄、开场白、分隔线 +2. 禁止输出 或任何 XML/JSON/代码块 +3. 所有【】节标题一个都不能少""" def parse_response(text): """从LLM回复中提取策略参数""" result = {"signal": "", "entry_low": 0, "entry_high": 0, "stop_loss": 0, "take_profit": 0, "position": ""} @@ -217,28 +326,44 @@ def parse_response(text): return result def save_result(code, full_text, parsed): - """保存LLM结果到DB""" + """保存LLM结果到DB（先快照再UPDATE）。空分析拒绝写入。""" + if not (full_text or "").strip(): + print(f" \u274c 拒绝写入空分析（LLM输出为空，保护已有数据）") + return conn = sqlite3.connect(DB) now = datetime.now().isoformat() + # ── 修改前快照 ── + snapshot_strategy_history(conn, code, 'batch_12d') + updates = ["full_analysis=?", "reassessed_at=?"] params = [full_text, now] if parsed["signal"]: updates.append("timing_signal=?") params.append(parsed["signal"]) - if parsed["entry_low"] > 0: + # 区间写入门禁：上下沿都必须为正且 下沿<上沿<下沿x3，否则视为解析错误整体跳过 + # （防 214.68~2.52 类解析污染，与 GATE_ZONE_SANITY 同级防护） + _el, _eh = parsed["entry_low"], parsed["entry_high"] + if _el > 0 and _eh > _el and _eh < _el * 3: updates.append("entry_low=?") - params.append(parsed["entry_low"]) - if parsed["entry_high"] > 0: + params.append(_el) updates.append("entry_high=?") - params.append(parsed["entry_high"]) - if parsed["stop_loss"] > 0: + params.append(_eh) + elif _el > 0 or _eh > 0: + print(f" ⚠️ 买入区解析异常({_el}~{_eh})，跳过区间写入（保留原值）", flush=True) + # 止损/止盈一致性门禁：损>0 时必须在区间下沿之下(0.5x~1.0x)，盈>0 时必须在区间上沿之上 + _sl, _tp = parsed["stop_loss"], parsed["take_profit"] + if _sl > 0 and (not _el or _sl < _el) and (not _tp or _sl < _tp): updates.append("stop_loss=?") - params.append(parsed["stop_loss"]) - if parsed["take_profit"] > 0: + params.append(_sl) + elif _sl > 0: + print(f" ⚠️ 止损{_sl}与区间/止盈不一致，跳过写入（保留原值）", flush=True) + if _tp > 0 and (not _eh or _tp > _eh) and (not _sl or _tp > _sl): updates.append("take_profit=?") - params.append(parsed["take_profit"]) + params.append(_tp) + elif _tp > 0: + print(f" ⚠️ 止盈{_tp}与区间/止损不一致，跳过写入（保留原值）", flush=True) if parsed["position"]: updates.append("position_advice=?") params.append(parsed["position"]) @@ -247,119 +372,179 @@ def save_result(code, full_text, parsed): sql = f"UPDATE holding_strategies SET {', '.join(updates)} WHERE code=? AND status='active'" conn.execute(sql, params) conn.commit() + + # ── 推荐操作 tag 同步（与 XMPP 动作级信号同源）── + sync_recommend_tag(conn, code, parsed.get("signal", "")) - # 买入信号→推XMPP通知（在conn close前执行） + # 买入信号→推XMPP通知（在conn close前执行）——推送质量门禁： + # 价格必须>0(live_prices实时价)、区间有效(下沿<上沿<下沿x3)、现价不超过上沿5%、 + # 损<下沿、盈>上沿、损在(0.5x~1.0x)现价内。任何一项不过 → 不推，只记日志。 if parsed.get("signal") == "买入": try: - _nr = conn.execute("SELECT name, price FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() + _nr = conn.execute("SELECT name FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() + _lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone() _name = _nr[0] if _nr else code - _p = _nr[1] if _nr else 0 + _p = _lp[0] if _lp and _lp[0] else 0 _el = parsed.get("entry_low", 0) _eh = parsed.get("entry_high", 0) _sl = parsed.get("stop_loss", 0) _tp = parsed.get("take_profit", 0) _pos = parsed.get("position", "") - _msg = f"📈 {_name}({code}) 价{_p}→12维分析生成买入信号！区间{_el}~{_eh} 损{_sl} 盈{_tp} 仓位{_pos}" - import urllib.request, json as _jj - _req = urllib.request.Request("http://127.0.0.1:5805/", - data=_jj.dumps({"body": _msg, "to": "hmo@yoin.fun", "type": "chat"}).encode(), - headers={"Content-Type": "application/json"}) - urllib.request.urlopen(_req, timeout=5) - print(f" 📨 XMPP推送成功: {_msg[:60]}") + _ok, _why = _validate_buy_alert(_p, _el, _eh, _sl, _tp) + if _ok: + _msg = f"📈 {_name}({code}) 价{_p}→12维分析生成买入信号！区间{_el}~{_eh} 损{_sl} 盈{_tp} 仓位{_pos}" + from alert_helper import notify as _notify, ACTION as _ACT + _notify("买入信号", _msg, _ACT) + print(f" \U0001f4e8 XMPP推送成功: {_msg[:60]}") + else: + print(f" ⚠️ 买入信号未过推送门禁({_why})，仅记日志不推送", flush=True) except Exception as _e: - print(f" ⚠️ XMPP推送失败: {_e}") + print(f" \u26a0\ufe0f XMPP推送失败: {_e}") conn.close() -def process_stock(code): + +def _validate_buy_alert(price, el, eh, sl, tp): + """买入信号推送门禁（垃圾信号不发）。 + 返回 (ok, reason)""" + if not price or price <= 0: + return False, f"无实时价格({price})" + if not (el > 0 and eh > el and eh < el * 3): + return False, f"区间无效({el}~{eh})" + if price > eh * 1.05: + return False, f"现价{price}高于区间上沿{eh}超5%（追高信号不推）" + if not (sl > 0 and sl < el and price * 0.5 <= sl <= price): + return False, f"止损{sl}不合理(需0.5x~1.0x现价且<下沿{el})" + if not (tp > eh and tp > sl): + return False, f"止盈{tp}需>上沿{eh}且>止损{sl}" + return True, "" + +def process_stock(code, force_today=False): """处理单只股票""" print(f"\n{'='*50}") print(f"处理: {code}") print(f"{'='*50}") - if has_llm_analysis(code): - print(f" ⏭ 已有LLM九维分析，跳过") + if in_cooldown(code): + print(f" \u23ed 冷却期内，跳过") return False - if in_cooldown(code): - print(f" ⏭ 冷却期内，跳过") + # 有分析且未过期 \u2192 跳过（除非 force_today 且今早未评） + if has_llm_analysis(code) and not analysis_stale(code, force_today): + print(f" \u23ed 已有12维分析且未过期，跳过") return False print(f" 收集数据...", flush=True) data = collect_data(code) if not data.get("price"): - print(f" ⚠️ 无价格数据，跳过") + print(f" \u26a0\ufe0f 无价格数据，跳过") return False print(f" 调LLM生成九维分析...", flush=True) prompt = build_prompt(data) - try: - r = subprocess.run(["curl", "-s", "--max-time", "300", - "-H", "Content-Type: application/json", - "-H", "Authorization: Bearer hermes123", - "-d", json.dumps({"model":"deepseek-v4-flash","messages":[{"role":"user","content":prompt}],"max_tokens":2048}), - GATEWAY], capture_output=True, timeout=310) - - if r.returncode != 0: - print(f" ❌ curl失败: {r.stderr.decode()[:100]}") - return False - - resp = json.loads(r.stdout) - if "choices" not in resp: - print(f" ❌ API异常: {str(resp)[:200]}") - return False - - full_text = resp["choices"][0]["message"]["content"] - print(f" ✅ LLM返回({len(full_text)}字)", flush=True) - - parsed = parse_response(full_text) - print(f" 信号={parsed['signal']} 区间={parsed['entry_low']}~{parsed['entry_high']} 损={parsed['stop_loss']} 盈={parsed['take_profit']} 仓位={parsed['position']}") - - save_result(code, full_text, parsed) - print(f" ✅ 已保存到DB") - return True - - except subprocess.TimeoutExpired: - print(f" ❌ 超时") - return False - except Exception as e: - print(f" ❌ 错误: {e}") + # ── 使用共享 LLM 客户端（替代 curl subprocess）── + result = call_llm(prompt, model=REASSESS_MODEL, max_tokens=4096) + + if not result["ok"] or not (result.get("content") or "").strip(): + print(f" \u274c LLM调用失败或空输出: {result.get('error') or 'empty content'}") return False + + full_text = result["content"] + print(f" \u2705 LLM返回({len(full_text)}字, {result['elapsed']:.1f}s, 尝试{result['attempts']}次)", flush=True) + + parsed = parse_response(full_text) + + # ── 截断保护：输出过短且无信号 = 低质输出，升级 pro 重试一次 ── + if not parsed.get("signal") and len(full_text) < 1500: + print(f" ⚠️ 输出截断({len(full_text)}字)且无信号，升级 {FALLBACK_MODEL} 重试...", flush=True) + result2 = call_llm(prompt, model=FALLBACK_MODEL, max_tokens=4096) + if result2["ok"] and len((result2.get("content") or "").strip()) > len(full_text): + full_text = result2["content"] + parsed = parse_response(full_text) + print(f" \u2705 升级后({len(full_text)}字)", flush=True) + + print(f" 信号={parsed['signal']} 区间={parsed['entry_low']}~{parsed['entry_high']} 损={parsed['stop_loss']} 盈={parsed['take_profit']} 仓位={parsed['position']}") + + save_result(code, full_text, parsed) + print(f" \u2705 已保存到DB") + return True def main(): + # ── 双通道预检：OCG直连 + hermes gateway 兜底，全挂才退出 ── + _ocg_ok = ocg_alive() + _gw_ok = gateway_alive() + if not _ocg_ok and not _gw_ok: + print("[FATAL] OCG上游与hermes gateway均不可用，退出") + sys.exit(1) + if not _ocg_ok: + print("[WARN] OCG直连不可用，将使用gateway兜底（agent运行时，较慢）") + if not _gw_ok: + print("[WARN] hermes gateway不可用，仅使用OCG直连") + codes = [] + force_today = "--today" in sys.argv + dtype = None + if "--type" in sys.argv: + idx = sys.argv.index("--type") + dtype = sys.argv[idx + 1] # holding | watchlist | all if "--code" in sys.argv: idx = sys.argv.index("--code") codes = [sys.argv[idx+1]] else: - # 所有自选策略 + # 按类型筛选 active 策略 + type_map = {"holding": "持仓策略", "watchlist": "自选策略"} conn = sqlite3.connect(DB) - rows = conn.execute("SELECT code FROM holding_strategies WHERE status='active' AND decision_type='自选策略' ORDER BY code").fetchall() + if dtype in type_map: + rows = conn.execute( + "SELECT code FROM holding_strategies WHERE status='active' AND decision_type=? ORDER BY code", + (type_map[dtype],)).fetchall() + else: + rows = conn.execute( + "SELECT code FROM holding_strategies WHERE status='active' ORDER BY decision_type, code").fetchall() conn.close() codes = [r[0] for r in rows] - print(f"待处理: {len(codes)}只") + print(f"待处理: {len(codes)}只 (type={dtype or 'all'}, force_today={force_today})") ok = 0 fail = 0 skip = 0 + failed_codes = [] for i, code in enumerate(codes): - if has_llm_analysis(code): - print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有LLM分析") + if has_llm_analysis(code) and not analysis_stale(code, force_today): + print(f" [{i+1}/{len(codes)}] \u23ed {code} 已有12维分析且未过期") skip += 1 continue print(f" [{i+1}/{len(codes)}] ", end="", flush=True) - if process_stock(code): + if process_stock(code, force_today): ok += 1 else: fail += 1 + failed_codes.append(code) - # 间隔15秒（防gateway过载） + # 间隔8秒（pro model较重但gateway可承受；retry逻辑吸收瞬断） if i < len(codes) - 1: - print(f" 等待15秒...", flush=True) - time.sleep(15) + print(f" 等待8秒...", flush=True) + time.sleep(8) + + # ── 失败二轮：主跑结束后休息 60s 让上游恢复，失败股整体重试一次 ── + # （凌晨上游空输出高发，二轮可救回大半；仍失败的留给下一轮调度） + if failed_codes: + print(f"\n{'='*50}") + print(f"失败二轮: {len(failed_codes)}只，休息60s后重试...") + time.sleep(60) + retry_ok = 0 + for code in failed_codes: + print(f" [retry] {code} ", end="", flush=True) + if process_stock(code, force_today): + retry_ok += 1 + ok += 1 + fail -= 1 + print(f" 等待8秒...", flush=True) + time.sleep(8) + print(f"失败二轮: {retry_ok}/{len(failed_codes)} 救回") print(f"\n{'='*50}") print(f"完成: {ok}成功, {fail}失败, {skip}跳过") diff --git a/archive/20260722-scripts-cleanup/candidate_filter.py b/archive/20260722-scripts-cleanup/candidate_filter.py index e33ab5b3..4a48c068 100644 --- a/archive/20260722-scripts-cleanup/candidate_filter.py +++ b/archive/20260722-scripts-cleanup/candidate_filter.py @@ -17,7 +17,9 @@ DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") UA = "Mozilla/5.0" def get_conn(): - return sqlite3.connect(str(DB_PATH)) + c = sqlite3.connect(str(DB_PATH), timeout=30) + c.execute("PRAGMA busy_timeout=30000") + return c def log_candidate(conn, code, stage, passed, detail): """记录过滤日志""" diff --git a/archive/20260722-scripts-cleanup/fix_gateway_port.py b/archive/20260722-scripts-cleanup/fix_gateway_port.py index 2d7b0996..283cc1b7 100644 --- a/archive/20260722-scripts-cleanup/fix_gateway_port.py +++ b/archive/20260722-scripts-cleanup/fix_gateway_port.py @@ -21,33 +21,26 @@ def port_open(port, host="127.0.0.1"): s.close() def check_session_health(): - """调gateway API，检测session是否卡死。超过15s无响应→不健康""" + """检测 gateway LLM 是否可用——扫 agent.log 最近一次真实调用结果。 + 不再发真实 LLM ping（25s 超时对 20-100s 的冷启动延迟必误报，且每次白烧 22k token）。 + """ try: - payload = json.dumps({ - "model": "hermes-agent", - "messages": [{"role": "user", "content": "ping"}] - }).encode() - req = urllib.request.Request(GATEWAY_URL, 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", SESSION_ID) - t0 = time.time() - with urllib.request.urlopen(req, timeout=25) as r: - data = json.loads(r.read()) - reply = data.get("choices", [{}])[0].get("message", {}).get("content", "") - elapsed = time.time() - t0 - if reply: - print(f"Session {SESSION_ID} 健康 ✓ ({elapsed:.1f}s)") - return True - else: - print(f"Session {SESSION_ID} 返回空", file=sys.stderr) - return False - except urllib.request.HTTPError as e: - print(f"Session {SESSION_ID} HTTP错误: {e.code}", file=sys.stderr) - return False + sys.path.insert(0, '/home/hmo/MoFin') + from xmpp_logger import _scan_agent_log + r = _scan_agent_log(time.time(), "zhiwei") + if r["status"] == "ok": + print(f"Session {SESSION_ID} 健康 ✓ (agent.log: latency={r.get('latency')}, {r.get('age_sec')}s前)") + return True + # error/unknown：只有近期有明确失败记录才判不健康 + if r["status"] == "error": + print(f"Session {SESSION_ID} 不健康: agent.log 最近调用失败 — {r.get('error','')[:100]}", file=sys.stderr) + return False + # unknown（无近期调用记录）= 空闲，不算不健康 + print(f"Session {SESSION_ID} 无近期调用记录（空闲正常）") + return True except Exception as e: - print(f"Session {SESSION_ID} 不健康: {e}", file=sys.stderr) - return False + print(f"Session {SESSION_ID} 健康检查异常: {e}（按健康处理）", file=sys.stderr) + return True def restart_gateway(): """通过systemd重启gateway""" diff --git a/archive/20260722-scripts-cleanup/mo_data.py b/archive/20260722-scripts-cleanup/mo_data.py new file mode 100644 index 00000000..cb168a13 --- /dev/null +++ b/archive/20260722-scripts-cleanup/mo_data.py @@ -0,0 +1,316 @@ +#!/usr/bin/env python3 +""" +mo_data.py — MoFin 统一数据层（纯 DB） + +所有数据从 SQLite 读取。不做 JSON fallback。 +JSON 文件已弃用，仅保留为历史备份。 + +用法: + from mo_data import read_portfolio, read_decisions, read_watchlist + + pf = read_portfolio() # 返回和 portfolio.json 一样的 dict 结构 + dec = read_decisions() # 返回和 decisions.json 一样的 dict 结构 + wl = read_watchlist() # 返回和 watchlist.json 一样的 dict 结构 +""" + +import sqlite3, json, sys +from datetime import datetime +from pathlib import Path + +DB_PATH = '/home/hmo/MoFin/data/mofin.db' +SCRIPT_DIR = Path('/home/hmo/MoFin/scripts') + + +def _get_db(): + db = sqlite3.connect(DB_PATH) + db.row_factory = sqlite3.Row + return db + + +# ── portfolio ───────────────────────────────────────────────────── + +def read_portfolio(): + """返回 portfolio.json 等价 dict。纯 DB。""" + db = _get_db() + rows = db.execute( + "SELECT code, name, shares, cost, price, market_value, " + "change_pct, currency, position_pct " + "FROM holdings WHERE is_active=1" + ).fetchall() + holdings = [] + for r in rows: + h = dict(r) + h['_currency'] = h.get('currency', 'CNY') + holdings.append(h) + + sum_row = db.execute("SELECT * FROM portfolio_summary WHERE id=1").fetchone() + summary = dict(sum_row) if sum_row else {} + + db.close() + + return { + "holdings": holdings, + "total_assets": summary.get("total_assets", 0), + "total_mv": summary.get("total_mv", 0), + "stock_value": summary.get("stock_value", summary.get("total_mv", 0)), + "cash": summary.get("cash", 0), + "frozen_cash": summary.get("frozen_cash", 0), + "position_pct": summary.get("position_pct", 0), + "currency": summary.get("currency", "CNY"), + "updated_at": summary.get("updated_at", ""), + } + + +# ── decisions ───────────────────────────────────────────────────── + +def _parse_json(val, default): + if val: + try: return json.loads(val) + except: pass + return default + + +def read_decisions(): + """返回 decisions.json 等价 dict。纯 DB。""" + db = _get_db() + rows = db.execute( + "SELECT code, name, version, price, cost, shares, " + "stop_loss, take_profit, entry_low, entry_high, " + "currency, strategy_type, action, timing_signal, " + "rr_ratio, tech_snapshot, stock_category, sector_context, " + "status, trigger_json, changelog_json, source, reason, " + "created_at, updated_at, " + "avg_price, decision_timestamp, note, quality_check, " + "quality_checked_at, quality_issues_json, position_advice, " + "signal_factors_json, time_horizon, decision_type, tag " + "FROM holding_strategies WHERE status IN ('active','updated') " + "ORDER BY code" + ).fetchall() + + decisions = [] + for r in rows: + d = dict(r) + d['trigger'] = _parse_json(r['trigger_json'], {}) + d['changelog'] = _parse_json(r['changelog_json'], []) + d['quality_issues'] = _parse_json(r['quality_issues_json'], {}) + d['signal_factors'] = _parse_json(r['signal_factors_json'], []) + d['timestamp'] = r['decision_timestamp'] or r['created_at'] or '' + d['type'] = r['decision_type'] or r['strategy_type'] or '持仓策略' + decisions.append(d) + + db.close() + + return { + "decisions": decisions, + "total": len(decisions), + "regenerated_at": datetime.now().strftime('%Y-%m-%d %H:%M'), + } + + +# ── watchlist ───────────────────────────────────────────────────── + +def read_watchlist(): + """返回 watchlist 等价 dict。纯 DB。 + 从 holding_strategies（自选策略）读取，watchlist_stocks 已废弃。""" + db = _get_db() + # 主数据源：holding_strategies 自选策略 + rows = db.execute( + "SELECT code, name, price, entry_low, entry_high, " + "stop_loss, currency, updated_at " + "FROM holding_strategies WHERE status='active' AND decision_type='自选策略'" + ).fetchall() + + stocks = [] + seen = set() + for r in rows: + code = str(r["code"]) + if code in seen: + continue + seen.add(code) + stocks.append({ + "code": code, + "name": r["name"] or "", + "price": r["price"] or 0, + "entry_low": r["entry_low"] or 0, + "entry_high": r["entry_high"] or 0, + "stop_loss": r["stop_loss"] or 0, + "currency": r["currency"] or "CNY", + "added_at": r["updated_at"] or "", + "analysis": {}, + }) + + return {"stocks": stocks, "total": len(stocks)} + + db.close() + + return { + "stocks": stocks, + "updated_at": datetime.now().strftime('%Y-%m-%d %H:%M'), + } + + +# ── 便捷别名 ─────────────────────────────────────────────────────── + +def read_portfolio_json(): + return read_portfolio() + +def read_decisions_json(): + return read_decisions() + +def read_watchlist_json(): + return read_watchlist() + + +# ── 统一价格获取（唯一入口，禁止各脚本自拉API）── + +def get_price(code, max_age_minutes=5, use_stale_fallback=True): + """获取单只股票最新价格。 + + 优先级: live_prices(DB) → stock_quote(API兜底) + - live_prices 有且不超过 max_age_minutes → 直接返回 + - 没有或过期 → 调 stock_quote 拉，写回 live_prices + - 都失败 → 返回 (None, None) + + 返回 (price, change_pct)，两值都是 float 或 None。 + """ + from mofin_db import get_price_from_db + from datetime import datetime, timedelta + + # 1. 先读 DB + try: + db_price, db_chg = get_price_from_db(code) + if db_price is not None and db_price > 0: + # 检查时效性 + conn = __import__('sqlite3').connect(str(DB_PATH)) + row = conn.execute( + "SELECT updated_at FROM live_prices WHERE code=?", + (str(code).strip(),) + ).fetchone() + conn.close() + if row and row[0]: + try: + updated = datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S") + age = (datetime.now() - updated).total_seconds() / 60 + if age <= max_age_minutes: + return (db_price, db_chg) + except: + pass + else: + return (db_price, db_chg) + except Exception: + pass + + # 2. DB 没有或过期 → 调 stock_quote + if not use_stale_fallback: + return (None, None) + + try: + import subprocess, json + r = subprocess.run( + [sys.executable, str(SCRIPT_DIR / "stock_quote.py"), str(code)], + capture_output=True, text=True, timeout=15 + ) + if r.returncode == 0: + data = json.loads(r.stdout.strip()) + price = float(data.get("price", 0)) + chg = float(data.get("change_pct", 0)) + if price > 0: + # 写回 live_prices + try: + conn = __import__('sqlite3').connect(str(DB_PATH)) + conn.execute(""" + INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) + VALUES (?, ?, ?, datetime('now','localtime')) + """, (str(code).strip(), price, chg)) + conn.commit() + conn.close() + except: + pass + return (price, chg) + except Exception: + pass + + return (None, None) + + +def get_prices_batch(codes, max_age_minutes=5): + """批量获取价格，返回 {code: (price, change_pct)}""" + from mofin_db import get_prices_batch_from_db + + result = {} + need_api = [] + + # 1. 批量读 DB + try: + db_prices = get_prices_batch_from_db(codes) + for code in codes: + cs = str(code).strip() + if cs in db_prices: + p, c = db_prices[cs] + if p and p > 0: + result[cs] = (p, c) + continue + need_api.append(cs) + except: + need_api = [str(c).strip() for c in codes] + + # 2. 缺失的调 API + if need_api: + try: + import subprocess, json + r = subprocess.run( + [sys.executable, str(SCRIPT_DIR / "stock_quote.py")] + need_api, + capture_output=True, text=True, timeout=30 + ) + if r.returncode == 0: + for line in r.stdout.strip().split("\n"): + if not line: + continue + try: + data = json.loads(line) + code = str(data.get("code", "")).strip() + price = float(data.get("price", 0)) + chg = float(data.get("change_pct", 0)) + if code and price > 0: + result[code] = (price, chg) + except: + pass + except: + pass + + return result + + +# ── cash_log 写入 ────────────────────────────────────────────────── + +def write_cash_log(cash_before, cash_after, frozen_before, frozen_after, + source, note, verified=0): + """记录现金变更到 cash_log 表。""" + change_amount = round(cash_after - cash_before, 2) if cash_after is not None and cash_before is not None else 0 + db = sqlite3.connect(DB_PATH) + try: + cur = db.execute( + """INSERT INTO cash_log + (timestamp, cash_before, cash_after, frozen_before, frozen_after, + change_amount, source, note, verified) + VALUES (datetime('now','localtime'), ?, ?, ?, ?, ?, ?, ?, ?)""", + (cash_before, cash_after, frozen_before, frozen_after, + change_amount, source, note, verified) + ) + db.commit() + return cur.lastrowid + finally: + db.close() + + +# ── 自检 ─────────────────────────────────────────────────────────── + +if __name__ == "__main__": + pf = read_portfolio() + print(f"portfolio: {len(pf.get('holdings',[]))} holdings, total_assets={pf.get('total_assets',0)}") + + dec = read_decisions() + print(f"decisions: {len(dec.get('decisions',[]))} entries") + + wl = read_watchlist() + print(f"watchlist: {len(wl.get('stocks',[]))} stocks") diff --git a/archive/20260722-scripts-cleanup/mofin_db.py b/archive/20260722-scripts-cleanup/mofin_db.py new file mode 100644 index 00000000..605e5b24 --- /dev/null +++ b/archive/20260722-scripts-cleanup/mofin_db.py @@ -0,0 +1,1426 @@ +#!/usr/bin/env python3 +"""mofin_db.py — MoFin 统一数据库访问层 + +所有脚本通过此模块访问 mofin.db，避免重复建表/连接逻辑。 + +用法: + from mofin_db import get_conn, write_market_snapshot, write_klines, ... + +设计原则: + - 幂等建表（CREATE TABLE IF NOT EXISTS） + - WAL 模式 + 外键约束 + - 所有写操作返回 (success: bool, detail: str) + - JSON 写入由调用方负责，本模块只写 SQLite +""" + +import sqlite3 +import json +import time +import functools +from datetime import datetime +from pathlib import Path +from typing import Optional, Callable + +DATA_DIR = Path("/home/hmo/MoFin/data") # 绝对路径：全系统唯一权威数据目录 +DB_PATH = DATA_DIR / "mofin.db" + +# ═══════════════════════════════════════════════════════════ +# 连接管理 +# ═══════════════════════════════════════════════════════════ + +def get_conn() -> sqlite3.Connection: + """获取数据库连接（WAL 模式，外键约束，Row 工厂，30秒超时防并发锁，autocommit模式）""" + DATA_DIR.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(DB_PATH), timeout=30, isolation_level=None) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + conn.execute("PRAGMA busy_timeout=30000") + conn.execute("PRAGMA synchronous=NORMAL") + # 每次连接时清理WAL：防止被kill的进程留下残留事务导致后续全部卡死 + try: + conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + except Exception: + pass + return conn + + +def execute_with_retry(conn: sqlite3.Connection, sql: str, params: tuple = (), + max_retries: int = 3, base_delay: float = 1.0) -> sqlite3.Cursor: + """执行SQL并自动重试（捕获 database is locked）""" + last_err = None + for attempt in range(max_retries + 1): + try: + return conn.execute(sql, params) + except sqlite3.OperationalError as e: + if "database is locked" not in str(e) and "cannot commit" not in str(e): + raise # 非锁错误直接抛 + last_err = e + if attempt < max_retries: + delay = base_delay * (2 ** attempt) # 指数退避: 1s, 2s, 4s + time.sleep(delay) + else: + raise sqlite3.OperationalError( + f"DB锁重试{max_retries}次仍失败: {e}" + ) + # unreachable -- both paths in loop either return or raise + if last_err: + raise last_err # type: ignore[misc] + + +def commit_with_retry(conn: sqlite3.Connection, max_retries: int = 3, + base_delay: float = 1.0) -> None: + """提交事务并自动重试""" + last_err = None + for attempt in range(max_retries + 1): + try: + conn.commit() + return + except sqlite3.OperationalError as e: + if "database is locked" not in str(e) and "cannot commit" not in str(e): + raise + last_err = e + if attempt < max_retries: + delay = base_delay * (2 ** attempt) + time.sleep(delay) + else: + raise sqlite3.OperationalError( + f"DB提交重试{max_retries}次仍失败: {e}" + ) + raise last_err + + +def retry_db_write(func: Callable) -> Callable: + """装饰器：为 DB 写函数自动添加重试""" + @functools.wraps(func) + def wrapper(*args, **kwargs): + max_retries = 3 + base_delay = 1.0 + last_err = None + for attempt in range(max_retries + 1): + try: + return func(*args, **kwargs) + except sqlite3.OperationalError as e: + if "database is locked" not in str(e) and "cannot commit" not in str(e): + raise + last_err = e + if attempt < max_retries: + delay = base_delay * (2 ** attempt) + time.sleep(delay) + else: + raise sqlite3.OperationalError( + f"DB写重试{max_retries}次仍失败({func.__name__}): {e}" + ) + raise last_err + return wrapper + + +# ═══════════════════════════════════════════════════════════ +# 建表（幂等） +# ═══════════════════════════════════════════════════════════ + +def init_all_tables(conn: sqlite3.Connection): + """创建全部表（幂等，已存在则跳过）""" + conn.executescript(""" + -- 市场快照 + CREATE TABLE IF NOT EXISTS market_snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp TEXT NOT NULL, + source TEXT NOT NULL DEFAULT 'ths', + up_ratio REAL, + mood TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')) + ); + CREATE INDEX IF NOT EXISTS idx_snapshots_time ON market_snapshots(timestamp); + + -- 板块快照 + CREATE TABLE IF NOT EXISTS sector_snapshots ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + snapshot_id INTEGER NOT NULL REFERENCES market_snapshots(id), + name TEXT NOT NULL, + change_pct REAL, + up_count INTEGER, + down_count INTEGER, + net_inflow REAL, + lead_stock TEXT, + lead_stock_change REAL, + volume REAL, + turnover REAL + ); + CREATE INDEX IF NOT EXISTS idx_sector_name ON sector_snapshots(name); + CREATE INDEX IF NOT EXISTS idx_sector_snapshot ON sector_snapshots(snapshot_id); + CREATE INDEX IF NOT EXISTS idx_sector_name_time ON sector_snapshots(name, snapshot_id); + + -- 个股 + CREATE TABLE IF NOT EXISTS stocks ( + code TEXT PRIMARY KEY, + name TEXT NOT NULL, + exchange TEXT DEFAULT 'SH', + type TEXT DEFAULT 'A', + updated_at TEXT + ); + + -- K线（日/周/月） + CREATE TABLE IF NOT EXISTS stock_daily ( + code TEXT NOT NULL REFERENCES stocks(code), + date TEXT NOT NULL, + open REAL, close REAL, high REAL, low REAL, + volume REAL, amount REAL, + PRIMARY KEY (code, date) + ); + CREATE TABLE IF NOT EXISTS stock_weekly ( + code TEXT NOT NULL REFERENCES stocks(code), + date TEXT NOT NULL, + open REAL, close REAL, high REAL, low REAL, + volume REAL, + PRIMARY KEY (code, date) + ); + CREATE TABLE IF NOT EXISTS stock_monthly ( + code TEXT NOT NULL REFERENCES stocks(code), + date TEXT NOT NULL, + open REAL, close REAL, high REAL, low REAL, + volume REAL, + PRIMARY KEY (code, date) + ); + + -- 基本面 + CREATE TABLE IF NOT EXISTS stock_fundamentals ( + code TEXT PRIMARY KEY REFERENCES stocks(code), + pe REAL, pb REAL, eps REAL, + mcap_total REAL, mcap_flow REAL, + updated_at TEXT + ); + + -- 板块成分映射 + CREATE TABLE IF NOT EXISTS stock_sectors ( + code TEXT NOT NULL REFERENCES stocks(code), + sector_name TEXT NOT NULL, + source TEXT DEFAULT 'ths', + updated_at TEXT DEFAULT (datetime('now','localtime')), + PRIMARY KEY (code, sector_name) + ); + CREATE INDEX IF NOT EXISTS idx_stock_sector ON stock_sectors(sector_name); + + -- 持仓 + CREATE TABLE IF NOT EXISTS holdings ( + code TEXT PRIMARY KEY REFERENCES stocks(code), + name TEXT NOT NULL, + shares INTEGER NOT NULL, + cost REAL, + price REAL, -- 当前价格 (CNY) + market_value REAL, -- 市值 = shares * price + change_pct REAL, -- 涨跌幅 + currency TEXT NOT NULL DEFAULT 'CNY' CHECK(currency IN ('CNY','HKD')), + position_pct REAL, + added_at TEXT, + is_active INTEGER DEFAULT 1, + closed_at TEXT, + close_pnl REAL + ); + + -- 持仓策略（对应 decisions.json decisions[]） + CREATE TABLE IF NOT EXISTS holding_strategies ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL REFERENCES holdings(code), + name TEXT, + version INTEGER DEFAULT 1, + price REAL, + cost REAL, + shares INTEGER DEFAULT 0, + stop_loss REAL, + take_profit REAL, + entry_low REAL, + entry_high REAL, + currency TEXT NOT NULL DEFAULT 'CNY' CHECK(currency IN ('CNY','HKD')), + strategy_type TEXT DEFAULT 'holding', + action TEXT, + timing_signal TEXT, + rr_ratio REAL, + tech_snapshot TEXT, + stock_category TEXT, + sector_context TEXT, + status TEXT DEFAULT 'active', + trigger_json TEXT, + changelog_json TEXT, + source TEXT, + reason TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')), + updated_at TEXT, + superseded_at TEXT, + -- 以下为 decisions.json→DB 迁移新增列 + avg_price REAL, + decision_timestamp TEXT, + note TEXT, + quality_check TEXT, + quality_checked_at TEXT, + quality_issues_json TEXT, + position_advice TEXT, + signal_factors_json TEXT, + time_horizon TEXT, + decision_type TEXT + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_strategy_code ON holding_strategies(code); + CREATE INDEX IF NOT EXISTS idx_strategy_status ON holding_strategies(status); + + -- 策略历史快照（每次覆写前自动记录） + CREATE TABLE IF NOT EXISTS strategy_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL, + name TEXT, + decision_type TEXT, + strategy_type TEXT, + full_analysis TEXT, + action TEXT, + timing_signal TEXT, + entry_low REAL, + entry_high REAL, + stop_loss REAL, + take_profit REAL, + position_advice TEXT, + rr_ratio REAL, + version INTEGER, + source_trigger TEXT, + reassessed_at TEXT, + snapshotted_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_strategy_history_code ON strategy_history(code, snapshotted_at); + + -- 自选股 + CREATE TABLE IF NOT EXISTS watchlist_stocks ( + code TEXT PRIMARY KEY REFERENCES stocks(code), + name TEXT NOT NULL, + price REAL, -- 当前价格 + entry_low REAL, -- 买入区下限 + entry_high REAL, -- 买入区上限 + stop_loss REAL, -- 止损 + currency TEXT NOT NULL DEFAULT 'CNY' CHECK(currency IN ('CNY','HKD')), + source TEXT, -- 来源: alpha_sift/xiaoguo/manual + source_detail TEXT, -- 来源详情 JSON + notes TEXT, -- 备注 + added_by TEXT, -- 谁加的 + added_at TEXT DEFAULT (datetime('now','localtime')), + is_active INTEGER DEFAULT 1, + analysis_json TEXT -- 分析结果 JSON + ); + + -- 候选池 + CREATE TABLE IF NOT EXISTS candidates ( + code TEXT PRIMARY KEY REFERENCES stocks(code), + name TEXT NOT NULL, + sector TEXT, + reason TEXT, + entry_range TEXT, + stop_loss REAL, + target REAL, + zhiwei_star REAL, + zhiwei_reviewed INTEGER DEFAULT 0, + zhiwei_reviewed_at TEXT, + promoted INTEGER DEFAULT 0, + promoted_at TEXT, + dropped INTEGER DEFAULT 0, + drop_reason TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')) + ); + + -- 候选评分历史 + CREATE TABLE IF NOT EXISTS candidate_score_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL REFERENCES candidates(code), + score REAL NOT NULL, + source TEXT NOT NULL, + reason TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')) + ); + CREATE INDEX IF NOT EXISTS idx_candidate_history ON candidate_score_history(code, created_at); + + -- 价格事件 + CREATE TABLE IF NOT EXISTS price_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL REFERENCES stocks(code), + name TEXT, + event_type TEXT NOT NULL, + price REAL, + trigger_value TEXT, + event_label TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')), + date TEXT + ); + CREATE INDEX IF NOT EXISTS idx_events_code ON price_events(code); + CREATE INDEX IF NOT EXISTS idx_events_date ON price_events(date); + + -- 策略评估记录 + CREATE TABLE IF NOT EXISTS strategy_evaluations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL REFERENCES stocks(code), + eval_type TEXT NOT NULL, + status TEXT DEFAULT 'pending', + old_stop_loss REAL, + new_stop_loss REAL, + old_tp REAL, + new_tp REAL, + reason TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')) + ); + + -- 持仓汇总（portfolio.json 顶层字段） + CREATE TABLE IF NOT EXISTS portfolio_summary ( + id INTEGER PRIMARY KEY CHECK (id = 1), + total_assets REAL, + total_mv REAL, -- 持仓总市值 + stock_value REAL, + cash REAL, -- 可用现金 + frozen_cash REAL DEFAULT 0, -- 冻结资金 + position_pct REAL, + total_pnl REAL, + currency TEXT NOT NULL DEFAULT 'CNY' CHECK(currency IN ('CNY','HKD')), + updated_at TEXT + ); + + -- 现金变更日志（每次买卖/出入金记录） + CREATE TABLE IF NOT EXISTS cash_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp TEXT NOT NULL DEFAULT (datetime('now','localtime')), + cash_before REAL, -- 变更前可用现金 + cash_after REAL, -- 变更后可用现金 + frozen_before REAL, -- 变更前冻结资金 + frozen_after REAL, -- 变更后冻结资金 + change_amount REAL, -- 现金变动额（正=入金/卖股，负=出金/买股） + source TEXT NOT NULL, -- 来源: screenshot/manual/import_xls/trade + note TEXT, -- 备注: 例如 "卖出法拉电子 200股" + verified INTEGER DEFAULT 0 -- 是否已验证（0=未验证，1=Dad确认） + ); + + -- 建议时间线（decisions.json advice_timeline[]） + CREATE TABLE IF NOT EXISTS advice_timeline ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL REFERENCES stocks(code), + date TEXT, + direction TEXT, + price REAL, + summary TEXT, + status TEXT, + evaluated INTEGER DEFAULT 0, + result TEXT, + evaluated_at TEXT, + report_id TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')) + ); + CREATE INDEX IF NOT EXISTS idx_advice_code ON advice_timeline(code); + + -- 准确率统计（accuracy_stats.json） + CREATE TABLE IF NOT EXISTS accuracy_stats ( + id INTEGER PRIMARY KEY CHECK (id = 1), + period_start TEXT, + period_end TEXT, + total_advice INTEGER DEFAULT 0, + correct INTEGER DEFAULT 0, + wrong INTEGER DEFAULT 0, + partial INTEGER DEFAULT 0, + unknown INTEGER DEFAULT 0, + pending INTEGER DEFAULT 0, + ignored INTEGER DEFAULT 0, + evaluated INTEGER DEFAULT 0, + accuracy_pct REAL, + phase1_correct INTEGER DEFAULT 0, + phase1_wrong INTEGER DEFAULT 0, + phase1_pending INTEGER DEFAULT 0, + phase1_accuracy REAL, + phase2_correct INTEGER DEFAULT 0, + phase2_wrong INTEGER DEFAULT 0, + phase2_pending INTEGER DEFAULT 0, + phase2_accuracy REAL, + total_evaluated INTEGER DEFAULT 0, + updated_at TEXT + ); + + -- 策略反馈（strategy_feedback.json） + CREATE TABLE IF NOT EXISTS strategy_feedback ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + code TEXT NOT NULL REFERENCES stocks(code), + name TEXT, + evaluated_at TEXT, + phase1_completed INTEGER DEFAULT 0, + phase1_result TEXT, + phase1_completed_at TEXT, + phase1_price REAL, + phase2_completed INTEGER DEFAULT 0, + phase2_result TEXT, + phase2_completed_at TEXT, + days_in_phase1 INTEGER, + adjustments_json TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')) + ); + CREATE INDEX IF NOT EXISTS idx_feedback_code ON strategy_feedback(code); + + -- 板块信号（trend_detector 产出） + CREATE TABLE IF NOT EXISTS sector_signals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + signal_type TEXT NOT NULL, + sector TEXT NOT NULL, + severity TEXT DEFAULT 'medium', + related_stocks TEXT, + holdings_in_sector TEXT, + watchlist_in_sector TEXT, + trigger_reason TEXT, + snapshot_id INTEGER, + processed INTEGER DEFAULT 0, + detected_at TEXT DEFAULT (datetime('now','localtime')) + ); + CREATE INDEX IF NOT EXISTS idx_signal_processed ON sector_signals(processed); + CREATE INDEX IF NOT EXISTS idx_signal_sector ON sector_signals(sector); + + -- 小果情报（xiaoguo_news_processor 产出） + CREATE TABLE IF NOT EXISTS signal_news ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + signal_id INTEGER REFERENCES sector_signals(id), + sector TEXT NOT NULL, + overall_sentiment TEXT, + summary TEXT, + key_articles TEXT, + searched_stocks TEXT, + created_at TEXT DEFAULT (datetime('now','localtime')) + ); + CREATE INDEX IF NOT EXISTS idx_signal_news_signal ON signal_news(signal_id); + + -- 小果扫描跟踪（去重用） + CREATE TABLE IF NOT EXISTS xiaoguo_scan_tracker ( + code TEXT PRIMARY KEY, + name TEXT, + last_scanned_at TEXT, + found_count INTEGER DEFAULT 0 + ); + + -- 实时价格快照（替代 live_prices.json） + CREATE TABLE IF NOT EXISTS live_prices ( + code TEXT PRIMARY KEY, + price REAL, + change_pct REAL, + updated_at TEXT DEFAULT (datetime('now','localtime')) + ); + + -- 多周期缓存（替代 multi_tf_cache.json） + CREATE TABLE IF NOT EXISTS mtf_cache ( + code TEXT PRIMARY KEY, + cache_json TEXT, + updated_at TEXT DEFAULT (datetime('now','localtime')) + ); + + -- 资金流缓存（替代 capital_flow_cache.json） + CREATE TABLE IF NOT EXISTS capital_flow_cache ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + cache_json TEXT, + updated_at TEXT DEFAULT (datetime('now','localtime')) + ); + + -- Self-TODO 自动化任务表 + CREATE TABLE IF NOT EXISTS todos ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + description TEXT, + status TEXT DEFAULT 'pending', + priority TEXT DEFAULT 'medium', + source TEXT DEFAULT 'manual', + fix_action TEXT, + retry_count INTEGER DEFAULT 0, + note TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + """) + conn.commit() + + # 迁移：给 signal_news 加 source 字段（幂等） + try: + conn.execute("ALTER TABLE signal_news ADD COLUMN source TEXT DEFAULT 'trend'") + except sqlite3.OperationalError: + pass + + # cash_log migration (2026-07-01) + try: + conn.execute("ALTER TABLE cash_log ADD COLUMN frozen_before REAL") + except sqlite3.OperationalError: + pass + try: + conn.execute("ALTER TABLE cash_log ADD COLUMN frozen_after REAL") + except sqlite3.OperationalError: + pass + try: + conn.execute("ALTER TABLE cash_log ADD COLUMN verified INTEGER DEFAULT 0") + except sqlite3.OperationalError: + pass + + # ── 币种约束迁移（2026-06-30）──────────────────────────────── + _currency_migrations = [ + ("holdings", ["price REAL", "market_value REAL", "change_pct REAL", + "currency TEXT NOT NULL DEFAULT 'CNY'"]), + ("holding_strategies", ["name TEXT", "price REAL", "cost REAL", "shares INTEGER DEFAULT 0", + "currency TEXT NOT NULL DEFAULT 'CNY'", + "action TEXT", "timing_signal TEXT", "rr_ratio REAL", + "tech_snapshot TEXT", "stock_category TEXT", + "sector_context TEXT", "status TEXT DEFAULT 'active'", + "trigger_json TEXT", "changelog_json TEXT", + "updated_at TEXT"]), + ("portfolio_summary", ["total_mv REAL", "frozen_cash REAL DEFAULT 0", + "currency TEXT NOT NULL DEFAULT 'CNY'"]), + ("watchlist_stocks", ["price REAL", "entry_low REAL", "entry_high REAL", + "stop_loss REAL", "currency TEXT NOT NULL DEFAULT 'CNY'", + "source TEXT", "source_detail TEXT", "notes TEXT", + "added_by TEXT", "analysis_json TEXT"]), + ] + for table, columns in _currency_migrations: + for col_def in columns: + col_name = col_def.split()[0] + try: + conn.execute(f"ALTER TABLE {table} ADD COLUMN {col_def}") + except sqlite3.OperationalError: + pass # column already exists + + # ── tag 迁移（2026-07-20）：推荐标签 current_recommend / active_manual ── + # 此前 strategy_lifecycle 在 dict 里设置 tag 但 write_holding_strategy 无此列， + # 导致标签在写入时被静默丢弃。补列 + 写入保留。 + try: + conn.execute("ALTER TABLE holding_strategies ADD COLUMN tag TEXT DEFAULT ''") + except sqlite3.OperationalError: + pass + conn.commit() + + +# ═══════════════════════════════════════════════════════════ +# 市场快照写入 +# ═══════════════════════════════════════════════════════════ + +def write_market_snapshot(conn: sqlite3.Connection, market_data: dict) -> tuple[bool, str, Optional[int]]: + """写入一次市场采集到 market_snapshots + sector_snapshots + + Returns: (ok, message, snapshot_id) + """ + try: + cur = conn.execute( + "INSERT INTO market_snapshots (timestamp, source, up_ratio, mood) VALUES (?, ?, ?, ?)", + (market_data["timestamp"], market_data.get("source", "unknown"), + market_data.get("up_ratio", 0), market_data.get("mood", "unknown")), + ) + sid = cur.lastrowid + + sectors = market_data.get("sectors", []) + rows = [(sid, s.get("name", ""), s.get("change", 0), + s.get("up_count"), s.get("down_count"), s.get("net_inflow"), + s.get("lead_stock"), s.get("lead_stock_change"), + s.get("volume"), s.get("turnover")) for s in sectors] + if rows: + conn.executemany( + "INSERT INTO sector_snapshots (snapshot_id, name, change_pct, up_count, down_count, " + "net_inflow, lead_stock, lead_stock_change, volume, turnover) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", rows) + conn.commit() + return True, f"snapshot_id={sid}, sectors={len(rows)}", sid + except Exception as e: + try: + conn.rollback() + except Exception: + pass + return False, str(e), None + + +# ═══════════════════════════════════════════════════════════ +# K线写入 +# ═══════════════════════════════════════════════════════════ + +def write_klines(conn: sqlite3.Connection, code: str, name: str, + daily: list = None, weekly: list = None, monthly: list = None, + fundamentals: dict = None) -> bool: + """将个股K线数据双写 SQLite + + Args: + code: 股票代码 + name: 股票名称 + daily/weekly/monthly: [{date, open, close, high, low, volume}, ...] + fundamentals: {pe, pb, eps, mcap_total, mcap_flow} + """ + try: + # 判断交易所 + raw = str(code) + if len(raw) == 5 and raw.isdigit(): + exchange, stype = "HK", "H" + elif raw.startswith(("6", "5", "9")): + exchange, stype = "SH", "A" + else: + exchange, stype = "SZ", "A" + + # stocks 表（INSERT OR REPLACE） + conn.execute( + "INSERT OR REPLACE INTO stocks (code, name, exchange, type, updated_at) VALUES (?, ?, ?, ?, ?)", + (code, name, exchange, stype, datetime.now().isoformat())) + + # K线数据 + for period, table, data in [ + ("daily", "stock_daily", daily), + ("weekly", "stock_weekly", weekly), + ("monthly", "stock_monthly", monthly), + ]: + if not data: + continue + rows = [(code, d.get("date", ""), d.get("open"), d.get("close"), + d.get("high"), d.get("low"), d.get("volume"), + d.get("amount") if period == "daily" else None) for d in data] + if period == "daily": + conn.executemany( + f"INSERT OR REPLACE INTO {table} (code, date, open, close, high, low, volume, amount) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", rows) + else: + conn.executemany( + f"INSERT OR REPLACE INTO {table} (code, date, open, close, high, low, volume) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + [(r[0], r[1], r[2], r[3], r[4], r[5], r[6]) for r in rows]) + + # 基本面 + if fundamentals: + conn.execute( + "INSERT OR REPLACE INTO stock_fundamentals (code, pe, pb, eps, mcap_total, mcap_flow, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (code, fundamentals.get("pe"), fundamentals.get("pb"), + fundamentals.get("eps"), fundamentals.get("mcap_total"), + fundamentals.get("mcap_flow"), datetime.now().isoformat())) + + conn.commit() + return True + except Exception as e: + try: + conn.rollback() + except Exception: + pass + return False + + +# ═══════════════════════════════════════════════════════════ +# 价格事件写入 +# ═══════════════════════════════════════════════════════════ + +def write_price_event(conn: sqlite3.Connection, code: str, name: str, + event_type: str, price: float, trigger_value: str, + event_label: str = "") -> bool: + """写入一条价格事件""" + try: + now = datetime.now() + conn.execute( + "INSERT INTO price_events (code, name, event_type, price, trigger_value, event_label, date) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (code, name, event_type, round(price, 2), trigger_value, + event_label, now.strftime("%Y-%m-%d"))) + conn.commit() + return True + except Exception: + try: + conn.rollback() + except Exception: + pass + return False + + +# ═══════════════════════════════════════════════════════════ +# 板块成分迁移 +# ═══════════════════════════════════════════════════════════ + +def migrate_stock_sectors(conn: sqlite3.Connection) -> tuple[int, int]: + """从 stock_sector_map.json 迁移到 stock_sectors 表 + + Returns: (migrated_stocks, total_mappings) + """ + sector_map_path = DATA_DIR / "stock_sector_map.json" + if not sector_map_path.exists(): + return 0, 0 + + try: + with open(sector_map_path, encoding="utf-8") as f: + data = json.load(f) + except Exception: + return 0, 0 + + # 过滤元数据字段 + mappings = [(code, sectors) for code, sectors in data.items() + if not code.startswith("_") and isinstance(sectors, list)] + + total = 0 + for code, sectors in mappings: + for sector in sectors: + try: + conn.execute( + "INSERT OR IGNORE INTO stock_sectors (code, sector_name, source) VALUES (?, ?, 'ths')", + (code, sector)) + total += 1 + except Exception: + pass + conn.commit() + return len(mappings), total + + +# ═══════════════════════════════════════════════════════════ +# 查询辅助 +# ═══════════════════════════════════════════════════════════ + +def query_sector_trend(conn: sqlite3.Connection, name: str, limit: int = 5) -> list[dict]: + """板块最近N次趋势""" + rows = conn.execute(""" + SELECT s.timestamp, ss.change_pct, ss.net_inflow, + ss.up_count, ss.down_count, ss.lead_stock, ss.lead_stock_change + FROM sector_snapshots ss + JOIN market_snapshots s ON ss.snapshot_id = s.id + WHERE ss.name = ? ORDER BY s.timestamp DESC LIMIT ? + """, (name, limit)).fetchall() + return [dict(r) for r in rows] + + +def query_top_inflow(conn: sqlite3.Connection, limit: int = 5) -> list[dict]: + """最新一次资金净流入排行""" + rows = conn.execute(""" + SELECT ss.name, ss.change_pct, ss.net_inflow, ss.lead_stock, s.timestamp + FROM sector_snapshots ss + JOIN market_snapshots s ON ss.snapshot_id = s.id + WHERE s.id = (SELECT MAX(id) FROM market_snapshots) + AND ss.net_inflow IS NOT NULL + ORDER BY ss.net_inflow DESC LIMIT ? + """, (limit,)).fetchall() + return [dict(r) for r in rows] + + +def query_consecutive_inflow(conn: sqlite3.Connection, days: int = 3) -> list[dict]: + """连续N次净流入的板块""" + rows = conn.execute(""" + SELECT name, COUNT(*) as times, ROUND(AVG(net_inflow), 2) as avg_inflow, + ROUND(AVG(change_pct), 2) as avg_change + FROM sector_snapshots ss + JOIN market_snapshots s ON ss.snapshot_id = s.id + WHERE s.id > (SELECT MAX(id) - ? FROM market_snapshots) + AND net_inflow > 0 + GROUP BY name HAVING COUNT(*) >= ? + ORDER BY avg_inflow DESC + """, (days, days)).fetchall() + return [dict(r) for r in rows] + + +def query_market_mood(conn: sqlite3.Connection, limit: int = 10) -> list[dict]: + """市场情绪趋势""" + rows = conn.execute(""" + SELECT timestamp, source, up_ratio, mood + FROM market_snapshots ORDER BY timestamp DESC LIMIT ? + """, (limit,)).fetchall() + return [dict(r) for r in rows] + + +def query_db_stats(conn: sqlite3.Connection) -> dict: + """数据库概览""" + snap_count = conn.execute("SELECT COUNT(*) FROM market_snapshots").fetchone()[0] + sector_count = conn.execute("SELECT COUNT(*) FROM sector_snapshots").fetchone()[0] + stock_count = conn.execute("SELECT COUNT(*) FROM stocks").fetchone()[0] + kline_count = conn.execute("SELECT COUNT(*) FROM stock_daily").fetchone()[0] + event_count = conn.execute("SELECT COUNT(*) FROM price_events").fetchone()[0] + holding_count = conn.execute("SELECT COUNT(*) FROM holdings").fetchone()[0] + candidate_count = conn.execute("SELECT COUNT(*) FROM candidates").fetchone()[0] + latest = conn.execute( + "SELECT timestamp, source FROM market_snapshots ORDER BY id DESC LIMIT 1").fetchone() + return { + "snapshots": snap_count, "sector_rows": sector_count, + "stocks": stock_count, "daily_klines": kline_count, + "price_events": event_count, "holdings": holding_count, + "candidates": candidate_count, + "latest_snapshot": dict(latest) if latest else None, + } + + +# ═══════════════════════════════════════════════════════════ +# 持仓查询 +# ═══════════════════════════════════════════════════════════ + +def query_holdings(conn: sqlite3.Connection) -> list[dict]: + """持仓列表（含最新策略）""" + rows = conn.execute(""" + SELECT h.code, h.name, h.shares, h.cost, h.position_pct, h.is_active, + h.price, h.change_pct, h.currency, + hs.stop_loss, hs.take_profit, hs.entry_low, hs.entry_high, + hs.reason as action, hs.created_at as strategy_updated + FROM holdings h + LEFT JOIN holding_strategies hs ON h.code = hs.code + AND hs.id = (SELECT MAX(id) FROM holding_strategies WHERE code = h.code AND strategy_type = 'holding') + WHERE h.is_active = 1 + """).fetchall() + return [dict(r) for r in rows] + + +def query_holding_by_code(conn: sqlite3.Connection, code: str) -> dict | None: + """单只持仓""" + row = conn.execute(""" + SELECT h.*, hs.stop_loss, hs.take_profit, hs.entry_low, hs.entry_high, + hs.reason as action + FROM holdings h + LEFT JOIN holding_strategies hs ON h.code = hs.code + AND hs.id = (SELECT MAX(id) FROM holding_strategies WHERE code = h.code AND strategy_type = 'holding') + WHERE h.code = ? + """, (code,)).fetchone() + return dict(row) if row else None + + +def query_portfolio_summary(conn: sqlite3.Connection) -> dict: + """持仓汇总""" + row = conn.execute("SELECT * FROM portfolio_summary WHERE id = 1").fetchone() + return dict(row) if row else {} + + +# ═══════════════════════════════════════════════════════════ +# 自选股查询 +# ═══════════════════════════════════════════════════════════ + +def query_watchlist(conn: sqlite3.Connection) -> list[dict]: + """自选股列表（含策略）""" + rows = conn.execute(""" + SELECT w.code, w.name, w.added_at, + hs.stop_loss, hs.take_profit, hs.entry_low, hs.entry_high, + hs.reason as action + FROM watchlist_stocks w + LEFT JOIN holding_strategies hs ON w.code = hs.code + AND hs.id = (SELECT MAX(id) FROM holding_strategies WHERE code = w.code AND strategy_type = 'watch') + WHERE w.is_active = 1 + """).fetchall() + return [dict(r) for r in rows] + + +# ═══════════════════════════════════════════════════════════ +# 决策/策略查询 +# ═══════════════════════════════════════════════════════════ + +def query_strategies(conn: sqlite3.Connection, code: str = None) -> list[dict]: + """策略列表（按版本倒序）""" + if code: + rows = conn.execute( + "SELECT * FROM holding_strategies WHERE code = ? ORDER BY version DESC", (code,)).fetchall() + else: + rows = conn.execute( + "SELECT * FROM holding_strategies ORDER BY code, version DESC").fetchall() + return [dict(r) for r in rows] + + +def query_advice_timeline(conn: sqlite3.Connection, code: str = None, limit: int = 50) -> list[dict]: + """建议时间线""" + if code: + rows = conn.execute( + "SELECT * FROM advice_timeline WHERE code = ? ORDER BY date DESC LIMIT ?", + (code, limit)).fetchall() + else: + rows = conn.execute( + "SELECT * FROM advice_timeline ORDER BY date DESC LIMIT ?", (limit,)).fetchall() + return [dict(r) for r in rows] + + +# ═══════════════════════════════════════════════════════════ +# 候选池查询 +# ═══════════════════════════════════════════════════════════ + +def query_candidates(conn: sqlite3.Connection, active_only: bool = True) -> list[dict]: + """候选池列表（含最新评分）""" + where = "WHERE c.dropped = 0" if active_only else "" + rows = conn.execute(f""" + SELECT c.*, (SELECT score FROM candidate_score_history + WHERE code = c.code ORDER BY created_at DESC LIMIT 1) as latest_score + FROM candidates c {where} + ORDER BY c.zhiwei_star DESC NULLS LAST + """).fetchall() + return [dict(r) for r in rows] + + +def query_candidate_scores(conn: sqlite3.Connection, code: str) -> list[dict]: + """某候选的评分历史""" + rows = conn.execute( + "SELECT * FROM candidate_score_history WHERE code = ? ORDER BY created_at", + (code,)).fetchall() + return [dict(r) for r in rows] + + +# ═══════════════════════════════════════════════════════════ +# 价格事件查询 +# ═══════════════════════════════════════════════════════════ + +def query_price_events(conn: sqlite3.Connection, code: str = None, limit: int = 100) -> list[dict]: + """价格事件""" + if code: + rows = conn.execute( + "SELECT * FROM price_events WHERE code = ? ORDER BY created_at DESC LIMIT ?", + (code, limit)).fetchall() + else: + rows = conn.execute( + "SELECT * FROM price_events ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall() + return [dict(r) for r in rows] + + +def query_price_events_by_date(conn: sqlite3.Connection, date: str) -> list[dict]: + """某天的价格事件""" + rows = conn.execute( + "SELECT * FROM price_events WHERE date = ? ORDER BY created_at DESC", (date,)).fetchall() + return [dict(r) for r in rows] + + +# ═══════════════════════════════════════════════════════════ +# 板块成分查询 +# ═══════════════════════════════════════════════════════════ + +def query_stock_sectors(conn: sqlite3.Connection, code: str) -> list[str]: + """某只股票所属板块""" + rows = conn.execute( + "SELECT sector_name FROM stock_sectors WHERE code = ?", (code,)).fetchall() + return [r[0] for r in rows] + + +def query_sector_stocks(conn: sqlite3.Connection, sector_name: str) -> list[str]: + """某板块包含的股票""" + rows = conn.execute( + "SELECT code FROM stock_sectors WHERE sector_name = ?", (sector_name,)).fetchall() + return [r[0] for r in rows] + + +# ═══════════════════════════════════════════════════════════ +# 准确率统计查询 +# ═══════════════════════════════════════════════════════════ + +def query_accuracy_stats(conn: sqlite3.Connection) -> dict: + """准确率统计""" + row = conn.execute("SELECT * FROM accuracy_stats WHERE id = 1").fetchone() + return dict(row) if row else {} + + +# ═══════════════════════════════════════════════════════════ +# 策略反馈查询 +# ═══════════════════════════════════════════════════════════ + +def query_strategy_feedback(conn: sqlite3.Connection, code: str = None) -> list[dict]: + """策略反馈""" + if code: + rows = conn.execute( + "SELECT * FROM strategy_feedback WHERE code = ? ORDER BY evaluated_at DESC", (code,)).fetchall() + else: + rows = conn.execute( + "SELECT * FROM strategy_feedback ORDER BY evaluated_at DESC").fetchall() + return [dict(r) for r in rows] + + +# ═══════════════════════════════════════════════════════════ +# 策略评估查询 +# ═══════════════════════════════════════════════════════════ + +def query_strategy_evaluations(conn: sqlite3.Connection, code: str = None) -> list[dict]: + """策略评估记录""" + if code: + rows = conn.execute( + "SELECT * FROM strategy_evaluations WHERE code = ? ORDER BY created_at DESC", (code,)).fetchall() + else: + rows = conn.execute( + "SELECT * FROM strategy_evaluations ORDER BY created_at DESC").fetchall() + return [dict(r) for r in rows] + + +# ═══════════════════════════════════════════════════════════ +# 市场快照查询（最新） +# ═══════════════════════════════════════════════════════════ + +def query_latest_market(conn: sqlite3.Connection) -> dict: + """获取最新一次市场快照（含 sector 详情）""" + row = conn.execute( + "SELECT * FROM market_snapshots ORDER BY id DESC LIMIT 1").fetchone() + if not row: + return {} + snap = dict(row) + # 关联 sectors + sectors = conn.execute( + "SELECT * FROM sector_snapshots WHERE snapshot_id = ? ORDER BY change_pct DESC", + (snap["id"],)).fetchall() + snap["sectors"] = [dict(r) for r in sectors] + snap["top_gainers"] = [dict(r) for r in sectors[:5]] + snap["top_losers"] = [dict(r) for r in sectors[-3:]] + return snap + + +# ═══════════════════════════════════════════════════════════════════ +# 通用工具 +# ═══════════════════════════════════════════════════════════════════ + +def get_price_from_db(code: str) -> tuple[float | None, float | None]: + """从 DB 读取最新价格（price_monitor 维护）。 + 返回 (price, change_pct) 或 (None, None) + + 所有脚本应优先调用此函数，DB 无数据时才拉腾讯 API。 + """ + try: + import sqlite3 + db = sqlite3.connect('/home/hmo/web-dashboard/data/mofin.db') + db.row_factory = sqlite3.Row + row = db.execute( + "SELECT price, change_pct FROM holdings WHERE code=? AND is_active=1", (str(code),) + ).fetchone() + if not row: + row = db.execute( + "SELECT price FROM holding_strategies WHERE code=? AND status='active' ORDER BY updated_at DESC LIMIT 1", (str(code),) + ).fetchone() + db.close() + if row: + return (row['price'], row['change_pct'] if 'change_pct' in row.keys() else None) + except Exception: + pass + return (None, None) + + +def get_prices_batch_from_db(codes: list[str]) -> dict: + """从 DB 批量读取价格。返回 {code: (price, change_pct)}""" + results = {} + if not codes: + return results + 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", (str(code),) + ).fetchone() + if not row: + row = db.execute( + "SELECT price FROM holding_strategies WHERE code=? AND status='active' ORDER BY updated_at DESC LIMIT 1", (str(code),) + ).fetchone() + if row and row['price']: + results[str(code)] = (row['price'], row['change_pct'] if 'change_pct' in row.keys() else 0) + db.close() + except Exception: + pass + return results + """最新一次市场快照（含板块数据）""" + snap = conn.execute( + "SELECT * FROM market_snapshots ORDER BY id DESC LIMIT 1").fetchone() + if not snap: + return {} + snap = dict(snap) + sectors = conn.execute( + "SELECT * FROM sector_snapshots WHERE snapshot_id = ? ORDER BY change_pct DESC", + (snap["id"],)).fetchall() + snap["sectors"] = [dict(r) for r in sectors] + # 计算 top_gainers / top_losers + snap["top_gainers"] = [dict(r) for r in sectors[:5]] + snap["top_losers"] = [dict(r) for r in sectors[-3:]] + return snap + + +# ═══════════════════════════════════════════════════════════════════ +# 核心写函数 — 替代 json.dump()，强制币种约束 +# ═══════════════════════════════════════════════════════════════════ + +def sync_recommend_tag(conn, code: str, timing_signal: str): + """裸 SQL 调用方（batch_reassess / per_stock_reassess）的推荐 tag 同步。 + 动作级信号 → current_recommend；信号降级 → 清除 current_recommend； + active_manual（人工标记）永不动。与 XMPP 动作级告警同源（红线#12）。""" + _ACTION_SIGNALS = ("买入", "可买入", "可加仓", "卖出", "止盈") + try: + if timing_signal in _ACTION_SIGNALS: + conn.execute( + "UPDATE holding_strategies SET tag='current_recommend' " + "WHERE code=? AND status='active' AND (tag IS NULL OR tag != 'active_manual')", + (code,)) + conn.commit() + elif timing_signal: + conn.execute( + "UPDATE holding_strategies SET tag='' " + "WHERE code=? AND status='active' AND tag='current_recommend'", + (code,)) + conn.commit() + except Exception as e: + print(f" [TAG SYNC] {code} 失败: {e}", flush=True) + + +def snapshot_strategy_history(conn, code: str, source_trigger: str = "write_holding_strategy"): + """在修改前快照当前策略到 strategy_history 表。永不抛异常。""" + try: + row = conn.execute( + "SELECT code, name, decision_type, strategy_type, full_analysis, " + "action, timing_signal, entry_low, entry_high, stop_loss, take_profit, " + "position_advice, rr_ratio, version, reassessed_at " + "FROM holding_strategies WHERE code=? AND status='active'", + (code,) + ).fetchone() + if not row: + return + now = datetime.now().isoformat() + conn.execute(""" + INSERT INTO strategy_history + (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) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, ( + row[0], row[1], row[2], row[3], + row[4], row[5], row[6], + row[7], row[8], row[9], row[10], + row[11], row[12], row[13], + source_trigger, row[14], now + )) + conn.commit() + # 每只股票只保留最近20条历史 + conn.execute(""" + DELETE FROM strategy_history WHERE code=? AND id NOT IN ( + SELECT id FROM strategy_history WHERE code=? ORDER BY snapshotted_at DESC LIMIT 20 + ) + """, (code, code)) + conn.commit() + except Exception as e: + print(f" [SNAPSHOT] {code} 快照失败: {e}", flush=True) + + +def write_holding_strategy(conn, code: str, name: str, data: dict, + source_trigger: str = "write_holding_strategy") -> tuple[bool, str]: + """写入持仓策略（替代 decisions.json 单条写入）。data 必须包含 currency。""" + try: + # ── 覆写前快照旧行 ── + snapshot_strategy_history(conn, code, source_trigger) + + + currency = data.get('currency', 'CNY') + # Serialize JSON fields + import json as _json + trigger_j = _json.dumps(data.get('trigger', {}), ensure_ascii=False) if isinstance(data.get('trigger'), dict) else str(data.get('trigger', '{}')) + changelog_j = _json.dumps(data.get('changelog', []), ensure_ascii=False) if isinstance(data.get('changelog'), list) else str(data.get('changelog', '[]')) + quality_issues_j = _json.dumps(data.get('quality_issues', {}), ensure_ascii=False) if isinstance(data.get('quality_issues'), dict) else data.get('quality_issues_json', '') + signal_factors_j = _json.dumps(data.get('signal_factors', []), ensure_ascii=False) if isinstance(data.get('signal_factors'), list) else data.get('signal_factors_json', '') + + # ── 推荐操作 tag 同步语义（与 XMPP 动作级信号同源，红线#12）── + # 动作级信号 → tag=current_recommend（进盯盘"推荐操作"区） + # 信号降级 → 清除 current_recommend（区域同步消失） + # active_manual（人工标记）永远不被自动流程覆盖或清除 + _ACTION_SIGNALS = ("买入", "可买入", "可加仓", "卖出", "止盈") + _existing_fa = data.get('full_analysis', '') + _existing_ra = data.get('reassessed_at', '') + _tag_absent = 'tag' not in data + _new_sig = data.get('timing_signal', '') or '' + _explicit_tag = data.get('tag', None) + _old_tag = '' + if not _existing_fa or _tag_absent: + try: + _old = conn.execute("SELECT full_analysis, reassessed_at, tag FROM holding_strategies WHERE code=? ORDER BY id DESC LIMIT 1", (code,)).fetchone() + if _old: + if not _existing_fa: + if _old[0]: _existing_fa = _old[0] + if _old[1]: _existing_ra = _old[1] + _old_tag = _old[2] or '' + except: + pass + if _old_tag == 'active_manual': + _existing_tag = 'active_manual' # 人工标记不可动 + elif _explicit_tag is not None: + _existing_tag = _explicit_tag # 显式传入优先（含''清除） + elif _new_sig in _ACTION_SIGNALS: + _existing_tag = 'current_recommend' # 动作级信号 → 自动推荐 + elif _new_sig and _old_tag == 'current_recommend': + _existing_tag = '' # 信号降级 → 清除自动推荐 + else: + _existing_tag = _old_tag # 其余保留 + + # ── 类型守卫：shares 必须是数值，防止字符串写入导致下游崩溃 ── + _shares = data.get('shares', 0) + if not isinstance(_shares, (int, float)): + print(f" [TYPE GUARD] {code} shares类型异常({type(_shares).__name__}={_shares!r})，重置为0", flush=True) + _shares = 0 + + # DELETE + INSERT + conn.execute("DELETE FROM holding_strategies WHERE code=?", (code,)) + conn.execute(""" + INSERT INTO holding_strategies + (code, name, version, price, cost, shares, stop_loss, take_profit, + entry_low, entry_high, currency, strategy_type, action, + timing_signal, rr_ratio, tech_snapshot, stock_category, + sector_context, status, trigger_json, changelog_json, + source, reason, updated_at, + avg_price, decision_timestamp, note, quality_check, + quality_checked_at, quality_issues_json, position_advice, + signal_factors_json, time_horizon, decision_type, + full_analysis, reassessed_at, tag) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, + datetime('now','localtime'), + ?,?,?,?,?,?,?,?,?,?,?,?,?) + """, ( + code, name, + data.get('version', 1), data.get('price'), data.get('cost'), + _shares, data.get('stop_loss'), data.get('take_profit'), + data.get('entry_low'), data.get('entry_high'), currency, + data.get('strategy_type', 'holding'), data.get('action'), + data.get('timing_signal'), data.get('rr_ratio'), + data.get('tech_snapshot'), data.get('stock_category'), + data.get('sector_context'), data.get('status', 'active'), + trigger_j, changelog_j, + data.get('source'), data.get('reason'), + # new columns + data.get('avg_price', 0), + data.get('timestamp') or data.get('created_at', ''), + data.get('note', ''), + data.get('quality_check', ''), + data.get('quality_checked_at', ''), + quality_issues_j, + data.get('position_advice', ''), + signal_factors_j, + data.get('time_horizon', ''), + data.get('type', data.get('strategy_type', 'holding')), + # 保留full_analysis和reassessed_at + _existing_fa, + _existing_ra, + _existing_tag, + )) + conn.commit() + return True, f"策略 {code} 已写入" + except sqlite3.IntegrityError as e: + return False, f"币种约束: {e}" + except Exception as e: + return False, str(e) + + +def write_holdings_batch(conn, holdings: list[dict]) -> tuple[bool, str]: + """批量写入持仓（替代 portfolio.json holdings[]）""" + try: + conn.execute("BEGIN IMMEDIATE") + for h in holdings: + currency = str(h.get('currency', 'CNY')).upper() + if currency not in ('CNY', 'HKD'): + return False, f"非法币种: {currency}（必须 CNY 或 HKD）" + conn.execute(""" + INSERT INTO holdings (code, name, shares, cost, price, market_value, + change_pct, currency, position_pct, added_at, is_active) + VALUES (?,?,?,?,?,?,?,?,?,datetime('now','localtime'),1) + ON CONFLICT(code) DO UPDATE SET + name=excluded.name, shares=excluded.shares, cost=excluded.cost, + price=excluded.price, market_value=excluded.market_value, + change_pct=excluded.change_pct, currency=excluded.currency, + position_pct=excluded.position_pct + """, ( + h.get('code'), h.get('name'), h.get('shares', 0), + h.get('cost'), h.get('price'), + h.get('market_value'), h.get('change_pct'), + h.get('currency', 'CNY'), h.get('position_pct'), + )) + conn.commit() + return True, f"已写入 {len(holdings)} 条持仓" + except sqlite3.IntegrityError as e: + conn.rollback() + return False, f"币种约束: {e}" + except sqlite3.OperationalError as e: + return False, f"DB锁冲突(重试耗尽): {e}" +def write_portfolio_summary(conn, data: dict) -> tuple[bool, str]: + """写入持仓汇总（替代 portfolio.json 顶层）""" + try: + conn.execute("BEGIN IMMEDIATE") + conn.execute(""" + INSERT INTO portfolio_summary (id, total_assets, total_mv, stock_value, + cash, frozen_cash, position_pct, total_pnl, currency, updated_at) + VALUES (1,?,?,?,?,?,?,?,?,datetime('now','localtime')) + ON CONFLICT(id) DO UPDATE SET + total_assets=excluded.total_assets, total_mv=excluded.total_mv, + stock_value=excluded.stock_value, cash=excluded.cash, + frozen_cash=excluded.frozen_cash, position_pct=excluded.position_pct, + total_pnl=excluded.total_pnl, currency=excluded.currency, + updated_at=datetime('now','localtime') + """, ( + data.get('total_assets'), data.get('total_mv'), data.get('stock_value'), + data.get('cash'), data.get('frozen_cash', 0), data.get('position_pct'), + data.get('total_pnl'), data.get('currency', 'CNY'), + )) + conn.commit() + return True, "汇总已写入" + except sqlite3.IntegrityError as e: + return False, f"约束: {e}" + except sqlite3.OperationalError as e: + return False, f"DB锁冲突: {e}" + + +def write_watchlist_stock(conn, stock: dict) -> tuple[bool, str]: + """写入自选股（写入 watchlist_stocks 表）""" + try: + conn.execute(""" + INSERT INTO watchlist_stocks (code, name, price, entry_low, entry_high, + stop_loss, currency, source, source_detail, notes, added_by, added_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,datetime('now','localtime')) + ON CONFLICT(code) DO UPDATE SET + name=excluded.name, price=excluded.price, entry_low=excluded.entry_low, + entry_high=excluded.entry_high, stop_loss=excluded.stop_loss, + currency=excluded.currency, source=excluded.source, + source_detail=excluded.source_detail, notes=excluded.notes, + added_by=excluded.added_by + """, ( + stock.get('code'), stock.get('name'), stock.get('price'), + stock.get('entry_low'), stock.get('entry_high'), stock.get('stop_loss'), + stock.get('currency', 'CNY'), stock.get('source'), stock.get('source_detail'), + stock.get('notes'), stock.get('added_by'), + )) + conn.commit() + return True, f"自选 {stock.get('code')} 已写入" + except sqlite3.IntegrityError as e: + return False, f"约束: {e}" + + +def write_cash_log(conn, data: dict) -> tuple[bool, str]: + """记录现金变更（替代手动改 portfolio.json cash 字段）""" + try: + conn.execute(""" + INSERT INTO cash_log (cash_before, cash_after, frozen_before, frozen_after, + change_amount, source, note) + VALUES (?,?,?,?,?,?,?) + """, ( + data.get('cash_before'), data.get('cash_after'), + data.get('frozen_before'), data.get('frozen_after'), + data.get('change_amount'), data.get('source', 'manual'), + data.get('note', ''), + )) + conn.commit() + return True, "现金变更已记录" + except Exception as e: + return False, str(e) + + +def query_cash_log(conn, limit: int = 20) -> list[dict]: + rows = conn.execute( + "SELECT * FROM cash_log ORDER BY id DESC LIMIT ?", (limit,) + ).fetchall() + return [dict(r) for r in rows] + + +# ═══ live_prices / mtf_cache / capital_flow_cache 写函数 ═══ + +def write_live_prices(conn, prices: dict): + """写入实时价格快照（替代 live_prices.json）""" + import json + for code, info in prices.items(): + conn.execute( + "INSERT OR REPLACE INTO live_prices (code, price, change_pct, updated_at) VALUES (?,?,?,datetime('now','localtime'))", + (code, info.get('price'), info.get('change_pct')) + ) + +def read_live_prices(conn) -> dict: + rows = conn.execute("SELECT code, price, change_pct FROM live_prices").fetchall() + return {r['code']: {'price': r['price'], 'change_pct': r['change_pct']} for r in rows} + + +def write_mtf_cache(conn, code: str, data: dict): + """写入多周期缓存（替代 multi_tf_cache.json 单条）""" + import json + conn.execute( + "INSERT OR REPLACE INTO mtf_cache (code, cache_json, updated_at) VALUES (?,?,datetime('now','localtime'))", + (code, json.dumps(data, ensure_ascii=False)) + ) + +def read_mtf_cache(conn, code: str) -> dict: + import json + r = conn.execute("SELECT cache_json FROM mtf_cache WHERE code=?", (code,)).fetchone() + return json.loads(r['cache_json']) if r else {} + + +def write_capital_flow_cache(conn, data: dict): + """写入资金流缓存（替代 capital_flow_cache.json）""" + import json + conn.execute("DELETE FROM capital_flow_cache") + conn.execute( + "INSERT INTO capital_flow_cache (cache_json, updated_at) VALUES (?,datetime('now','localtime'))", + (json.dumps(data, ensure_ascii=False),) + ) + +def read_capital_flow_cache(conn) -> dict: + import json + r = conn.execute("SELECT cache_json FROM capital_flow_cache ORDER BY id DESC LIMIT 1").fetchone() + return json.loads(r['cache_json']) if r else {} diff --git a/archive/20260722-scripts-cleanup/morning_health_check.py b/archive/20260722-scripts-cleanup/morning_health_check.py old mode 100755 new mode 100644 index 89ca01f7..dd89543d --- a/archive/20260722-scripts-cleanup/morning_health_check.py +++ b/archive/20260722-scripts-cleanup/morning_health_check.py @@ -42,15 +42,12 @@ HERMES_CRON_DIR = Path("/home/hmo/.hermes/profiles/position-analyst/cron") def derive_fix_action(detail, msg): """根据issue信息推导可执行的修复命令""" - # 小果扫描 error → 验证脚本是否存在 - if "xiaoguo_scanner" in msg or "小果扫描" in msg: - return f"ls -la /home/hmo/.hermes/profiles/position-analyst/scripts/xiaoguo_scanner.py 2>&1 && echo 'ok'" # system-audit error → 验证拷贝 if "system_audit" in msg or "系统审计" in msg: return f"ls -la /home/hmo/.hermes/profiles/position-analyst/scripts/system_audit.py 2>&1" # cron errors（last_status=error）→ 验证文件存在，等下次cron运行自动恢复 - if "cron" in msg.lower() and "error" in msg.lower() and ("小果" in msg or "系统审计" in msg): - return f"ls -la /home/hmo/.hermes/profiles/position-analyst/scripts/xiaoguo_scanner.py /home/hmo/.hermes/profiles/position-analyst/scripts/system_audit.py 2>&1" + if "cron" in msg.lower() and "error" in msg.lower() and "系统审计" in msg: + return f"ls -la /home/hmo/.hermes/profiles/position-analyst/scripts/system_audit.py 2>&1" # 港股汇率 → 刷新 if "港股汇率" in msg: return f"cd {BASE} && python3 hk_rate.py 2>&1" @@ -60,9 +57,9 @@ def derive_fix_action(detail, msg): # delivery目标缺失 → 改为local if "deliver" in msg.lower() or "delivery" in msg.lower(): return f"cd {BASE} && echo '需手动设置: cronjob action=update deliver=local'" - # 小果→知微桥不通 + # 小果→知微桥不通（小果已归档，不再自动修复） if "信号桥" in msg: - return f"cd {BASE} && python3 scripts/xiaoguo_signal_consumer.py 2>&1" + return None # 小果已归档，信号桥不再使用 return None @@ -319,8 +316,6 @@ def check_db_table_count(table, field, value, op="today", threshold=0): cols = [r[1] for r in cur.execute(f"PRAGMA table_info({table})").fetchall()] if "processed" in cols: sql = f"SELECT COUNT(*) FROM {table} WHERE (processed = 0 OR processed IS NULL)" - elif "source" in cols: - sql = f"SELECT COUNT(*) FROM {table} WHERE source LIKE '%xiaoguo%'" else: sql = f"SELECT COUNT(*) FROM {table}" cur.execute(sql) @@ -400,7 +395,6 @@ def check_cron_paused(): """检查不应暂停的cron是否被误暂停""" should_run = [ ("3a9fb3300a6a", "价格监控"), - ("0851c7838ca3", "小果扫描"), ("e13323928f3a", "自选提醒"), ("b809fcabfa5b", "分支评估"), ] @@ -604,10 +598,9 @@ def run_check(item): elif check_spec == "meta:checklist_completeness": ok, detail = check_meta_checklist_completeness() elif check_spec == "pipeline:xiaoguo_signal_flow": - today_xiaoguo, d1 = check_db_table_count("signal_news", "created_at", None, "today", 0) - unproc, d2 = check_db_table_count("signal_news", None, None, "unprocessed", 30) - ok = today_xiaoguo or unproc - detail = f"today_xiaoguo={d1}, unprocessed={d2}" + # 小果已归档，此管道不再检查 + ok = True + detail = "小果已归档，跳过信号流检查" elif check_spec == "pipeline:registry_audit": ok = True gaps = [] diff --git a/archive/20260722-scripts-cleanup/per_stock_reassess.py b/archive/20260722-scripts-cleanup/per_stock_reassess.py index c5c1d172..b84fd8fc 100644 --- a/archive/20260722-scripts-cleanup/per_stock_reassess.py +++ b/archive/20260722-scripts-cleanup/per_stock_reassess.py @@ -34,8 +34,11 @@ def _in_cooldown(code): sys.path.insert(0, "/home/hmo/web-dashboard") sys.path.insert(0, "/home/hmo/MoFin") +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # profile-scripts 硬链目录 from strategy_lifecycle import reassess_with_context as reassess_strategy from mo_data import read_decisions, read_portfolio +from llm_client import call_llm, REASSESS_MODEL +from mofin_db import snapshot_strategy_history def _build_full_analysis(code, entry, result): @@ -431,18 +434,76 @@ def main(): except: pass - _prompt = f"""你是一个资深股票分析师。请对股票{code}做一个完整的12维矩阵分析（3横×4纵：大盘/行业/个股 × 基本面/消息面/技术面/资金面）。 + # ── 拉取已有策略全文 + 最近变更 ── + _existing_full_analysis = "" + _existing_changelog_text = "无变更记录" + try: + _edb = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") + _er = _edb.execute( + "SELECT full_analysis, changelog_json FROM holding_strategies " + "WHERE code=? AND status='active'", (code,) + ).fetchone() + if _er: + _existing_full_analysis = _er[0] or "" + _cl_raw = _er[1] or "" + if _cl_raw: + _cl = __import__('json').loads(_cl_raw) if isinstance(_cl_raw, str) else _cl_raw + if isinstance(_cl, list) and _cl: + _recent = _cl[-3:] + _existing_changelog_text = "\n".join( + [f" [{c.get('timestamp','?')}] {c.get('action','?')}: {c.get('reason','')}"[:120] + for c in reversed(_recent)] + ) + _edb.close() + except: + pass + + _prompt = f"""你是一个资深股票分析师。请对股票{code}评估现有策略是否仍然有效，并输出完整的新策略。 + +╔══════════════════════════════════════════════╗ +║ 📋 第一步：审阅原策略 ║ +╚══════════════════════════════════════════════╝ + +【原策略全文】（上次完整分析）： +{_existing_full_analysis or '暂无完整策略分析'} + +【当前策略参数】： + 价格={price} 信号={result.get("timing_signal") or entry.get("timing_signal","")} + 买入区间={entry.get("entry_low",0)}~{entry.get("entry_high",0)} + 止损={entry.get("stop_loss",0)} 止盈={entry.get("take_profit",0)} + RR={result.get("rr_ratio", entry.get("rr_ratio", 0))} + 策略={result.get("action") or entry.get("action","")} + 行业={(result.get("sector_context") or entry.get("sector_context",""))[:50]}（当日实时） + 技术={(result.get("tech_snapshot") or entry.get("tech_snapshot",""))[:200]}（MA=5/10/20/60日 支撑阻力=近20日 量价=当日+近5日趋势） + +【最近变更记录】： +{_existing_changelog_text} + +╔══════════════════════════════════════════════╗ +║ 📊 第二步：12维矩阵交叉分析 ║ +╚══════════════════════════════════════════════╝ ⚠️ 重要：12个维度必须交叉对比，找出矛盾/共振点，给出综合判断。 -当前数据（实时API，每条标注时间窗口，禁止使用模型训练数据）： -大盘={_macro_desc or "震荡"}（当日实时） | PE/市值={_pe_val} {_pb_val}（最新财报） | 价格={price} 区间={entry.get("entry_low",0)}~{entry.get("entry_high",0)} 止损={entry.get("stop_loss",0)} 止盈={entry.get("take_profit",0)} RR={result.get("rr_ratio",entry.get("rr_ratio",0))} | 信号={result.get("timing_signal") or entry.get("timing_signal","")} | 行业={(result.get("sector_context") or entry.get("sector_context",""))[:50]}（当日实时） -策略={(result.get("action") or entry.get("action",""))[:200]} -技术={(result.get("tech_snapshot") or entry.get("tech_snapshot",""))[:200]}（MA=5/10/20/60日 支撑阻力=近20日 量价=当日+近5日趋势） +当前实时数据（每条标注时间窗口，禁止使用模型训练数据）： +大盘={_macro_desc or "震荡"}（当日实时） | PE/市值={_pe_val} {_pb_val}（最新财报） 资金流={_flow_note}（近5日累计） 消息面={_news_note}（最近3条，自动标注抓取时间） -格式： +╔══════════════════════════════════════════════╗ +║ 📝 第三步：决策输出 ║ +╚══════════════════════════════════════════════╝ + +请严格按以下顺序输出： + +【维持或修改】判断当前策略是否仍然有效，回答「维持」或「修改」。 + +【修改点及理由】（如果维持，写「无需修改」；如果修改，逐条列出）： + - 修改什么参数/方向 + - 理由（引用具体维度矛盾或共振） + +【最终新策略】（完整策略全文，self-contained，可直接存入DB） + 【交叉分析】哪些维度矛盾/共振，关键信号 ① 大盘×基本面 ② 大盘×消息面 ③ 大盘×技术面 ④ 大盘×资金面 ⑤ 行业×基本面 ⑥ 行业×消息面 ⑦ 行业×技术面 ⑧ 行业×资金面 @@ -452,26 +513,42 @@ def main(): 【综合结论】(买入/关注/观望/卖出) 【操作建议】 【建议止损】 -【建议止盈】""" - try: - _ur = __import__('urllib.request', fromlist=['Request']) - _req = _ur.Request("http://127.0.0.1:8643/v1/chat/completions", - data=__import__('json').dumps({"model":"deepseek-v4-flash","messages":[{"role":"user","content":_prompt}],"max_tokens":1024}).encode(), - headers={"Content-Type":"application/json","Authorization":"Bearer hermes123"}) - _resp = _ur.build_opener(_ur.ProxyHandler({})).open(_req, timeout=300) - _llm_out = __import__('json').loads(_resp.read().decode())["choices"][0]["message"]["content"] - _full_analysis_text = _llm_out - print(f" ✅ LLM12维分析完成({len(_full_analysis_text)}字)", flush=True) - except Exception as _e: - print(f" ❌ LLM12维分析失败: {_e}", file=__import__('sys').stderr) - _full_analysis_text = None +【建议止盈】 +【建议仓位】⚠️不可省略，非"买入"时写"不新建仓" - # 保存到DB +⚠️ 输出纪律（必须遵守）： +1. 直接以【维持或修改】开头，禁止任何寒暄、开场白、分隔线 +2. 禁止输出 或任何 XML/JSON/代码块 +3. 所有【】节标题一个都不能少""" + _full_analysis_text = None + try: + _llm_result = call_llm(_prompt, max_tokens=4096, timeout=150, retries=1, backoff=20) + if _llm_result["ok"]: + _full_analysis_text = _llm_result["content"] + print(f" ✅ LLM12维分析完成({len(_full_analysis_text)}字, {_llm_result['elapsed']:.1f}s)", flush=True) + else: + print(f" ❌ LLM12维分析失败({_llm_result['attempts']}次): {_llm_result['error'][:200]}", flush=True) + except Exception as _e: + print(f" ❌ LLM12维分析异常: {_e}", flush=True) + + # ── 保存到DB（覆写前先快照）── _fa_conn = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") - _fa_conn.execute("UPDATE holding_strategies SET full_analysis=?, reassessed_at=? WHERE code=? AND status='active'", (_full_analysis_text, __import__('datetime').datetime.now().isoformat(), code)) - _fa_conn.commit() + if _full_analysis_text: + # 快照旧策略（使用共享函数） + try: + snapshot_strategy_history(_fa_conn, code, "per_stock_12d") + except Exception as _se: + print(f" ⚠️ 快照失败: {_se}", flush=True) + + _fa_conn.execute( + "UPDATE holding_strategies SET full_analysis=?, reassessed_at=? WHERE code=? AND status='active'", + (_full_analysis_text, __import__('datetime').datetime.now().isoformat(), code)) + _fa_conn.commit() _fa_conn.close() - print(f" ✅ 完整12维分析已保存({len(_full_analysis_text)}字)" if _full_analysis_text else f" ⚠️ 12维分析未完成，跳过保存") + if _full_analysis_text: + print(f" ✅ 完整12维分析已保存({len(_full_analysis_text)}字)") + else: + print(f" ⚠️ 12维分析未完成，跳过保存") print(f" [DB] holding_strategies 已更新: {code}") # 从LLM输出提取信号 if _full_analysis_text and '【综合结论】' in _full_analysis_text: @@ -480,8 +557,14 @@ def main(): if _sig_line: _sig = '买入' if '买入' in _sig_line[0] else '关注' if '关注' in _sig_line[0] else '观望' if '观望' in _sig_line[0] else '卖出' if '卖出' in _sig_line[0] else '' if _sig: - __import__('sqlite3').connect('/home/hmo/MoFin/data/mofin.db').execute( - "UPDATE holding_strategies SET timing_signal=? WHERE code=? AND status='active'", (_sig, code)).connection.commit() + _ts_conn = __import__('sqlite3').connect('/home/hmo/MoFin/data/mofin.db') + _ts_conn.execute( + "UPDATE holding_strategies SET timing_signal=? WHERE code=? AND status='active'", (_sig, code)) + _ts_conn.commit() + # 推荐操作 tag 同步（与 XMPP 动作级信号同源） + from mofin_db import sync_recommend_tag + sync_recommend_tag(_ts_conn, code, _sig) + _ts_conn.close() print(f" ✅ LLM信号={_sig} 已写入") # 买入信号→推XMPP if _sig == "买入": @@ -490,10 +573,8 @@ def main(): "SELECT name, price, entry_low, entry_high, stop_loss, take_profit, position_advice FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() if _nr2: _xm = f"📈 {_nr2[0] or code}({code}) 价{_nr2[1]}→12维买入信号！区间{_nr2[2]}~{_nr2[3]} 损{_nr2[4]} 盈{_nr2[5]} 仓位{_nr2[6] or '-'}" - _xr = __import__('urllib.request').Request("http://127.0.0.1:5805/", - data=__import__('json').dumps({"body": _xm, "to": "hmo@yoin.fun", "type": "chat"}).encode(), - headers={"Content-Type": "application/json"}) - __import__('urllib.request').urlopen(_xr, timeout=5) + from alert_helper import notify as _notify2, ACTION as _ACT2 + _notify2("买入信号", _xm, _ACT2) print(f" 📨 XMPP推送买入信号") except: pass except: pass diff --git a/archive/20260722-scripts-cleanup/price_monitor.py b/archive/20260722-scripts-cleanup/price_monitor.py index b5159d11..25f5cd92 100644 --- a/archive/20260722-scripts-cleanup/price_monitor.py +++ b/archive/20260722-scripts-cleanup/price_monitor.py @@ -608,7 +608,16 @@ def run_once(round_label=""): # === 第三步：买入区偏离检测 + 自动重评 === reassesed_codes = [] # 先做急跌检测（仅持仓，自选股不推送暴跌告警） - holdings_codes = {d["code"] for d in active if (d.get("shares") or 0) > 0} + holdings_codes = set() + for d in active: + shares = d.get("shares", 0) + if isinstance(shares, (int, float)): + if shares > 0: + holdings_codes.add(d["code"]) + else: + # 非数值shares（如被错误写入的字符串），兜底处理 + holdings_codes.add(d["code"]) + print(f" [WARN] {d.get('code')} shares为非数值({shares!r})，视为持仓处理", flush=True) for d in active: code = d["code"] # 非持仓跳过 diff --git a/archive/20260722-scripts-cleanup/stale_detector.py b/archive/20260722-scripts-cleanup/stale_detector.py index 93412551..fc805b83 100644 --- a/archive/20260722-scripts-cleanup/stale_detector.py +++ b/archive/20260722-scripts-cleanup/stale_detector.py @@ -142,7 +142,7 @@ def main(): reassess_scripts.append(code) print(f"[AUTO_REASSESS] {name}({code}) 价{cur_price:.2f}偏离买入区中心{center:.2f} {drift:+.0f}% → 触发重评") if reassess_scripts: - # 调用 per_stock_reassess + # 调用 per_stock_reassess（每轮最多5只，防LLM慢导致整批超时；其余下轮继续） reassess_path = None for p in ['/home/hmo/MoFin/scripts/per_stock_reassess.py', '/home/hmo/.hermes/profiles/position-analyst/scripts/per_stock_reassess.py']: @@ -150,12 +150,20 @@ def main(): reassess_path = p break if reassess_path: - for code in reassess_scripts: - r = subprocess.run(['python3', reassess_path, code], - capture_output=True, text=True, timeout=60) - out = r.stdout.strip()[:200] if r.stdout else "" - err = r.stderr.strip()[:200] if r.stderr else "" - print(f" → {code}: exited={r.returncode} {out}") + MAX_PER_RUN = 5 + batch = reassess_scripts[:MAX_PER_RUN] + if len(reassess_scripts) > MAX_PER_RUN: + print(f"[AUTO_REASSESS] 本轮限{MAX_PER_RUN}只，剩余{len(reassess_scripts)-MAX_PER_RUN}只下轮继续") + for code in batch: + try: + # LLM 重评冷启动 20-100s，deepseek-v4-pro 更慢 → 480s + r = subprocess.run(['python3', reassess_path, code], + capture_output=True, text=True, timeout=480) + out = r.stdout.strip()[:200] if r.stdout else "" + err = r.stderr.strip()[:200] if r.stderr else "" + print(f" → {code}: exited={r.returncode} {out}") + except subprocess.TimeoutExpired: + print(f" → {code}: 超时480s（LLM仍慢），下轮重试") except Exception as e: print(f"[AUTO_REASSESS FAIL] {e}") # ----- 结束 自选股重评 ----- diff --git a/archive/20260722-scripts-cleanup/verify_reassess_pipeline.py b/archive/20260722-scripts-cleanup/verify_reassess_pipeline.py index 464f5b91..0e99fd4d 100644 --- a/archive/20260722-scripts-cleanup/verify_reassess_pipeline.py +++ b/archive/20260722-scripts-cleanup/verify_reassess_pipeline.py @@ -94,7 +94,21 @@ def run(): # 1. price_monitor 最近运行时间 try: - conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") + conn = None + last_err = None + # malformed 可能是 I/O 风暴下的瞬态 WAL 损坏（2026-07-21 事件）： + # checkpoint 后自愈。重试一次再告警，避免误报轰炸 + for _attempt in range(2): + try: + conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db") + conn.execute("SELECT 1 FROM live_prices LIMIT 1").fetchone() + break + except Exception as e: + last_err = e + import time as _t + _t.sleep(3) + if conn is None: + raise last_err lp = conn.execute("SELECT MAX(updated_at) FROM live_prices").fetchone()[0] if lp: lp_dt = datetime.fromisoformat(lp) if isinstance(lp, str) else lp