2391 lines
117 KiB
Python
2391 lines
117 KiB
Python
#!/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);
|
||
|
||
-- 策略追踪评估（2026-07-27 老爸：每条推荐操作的完整生命周期跟踪）
|
||
-- 每个版本一条记录，策略变更时自动追加新版本
|
||
CREATE TABLE IF NOT EXISTS strategy_tracking (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
code TEXT NOT NULL,
|
||
name TEXT,
|
||
version_seq INTEGER DEFAULT 1, -- 该股票的第几个策略版本
|
||
-- 推荐时的快照
|
||
tracked_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
|
||
timing_signal TEXT,
|
||
rec_score INTEGER DEFAULT 0,
|
||
rr_ratio REAL,
|
||
entry_low REAL,
|
||
entry_high REAL,
|
||
entry_mid REAL,
|
||
stop_loss REAL,
|
||
take_profit REAL,
|
||
position_advice TEXT,
|
||
price_at_track REAL, -- 记录时的市价
|
||
-- 区别于前一版本的变化摘要
|
||
change_summary TEXT,
|
||
-- 结果跟踪
|
||
status TEXT DEFAULT 'active' CHECK(status IN ('active','hit_tp','hit_sl','expired','manual_close')),
|
||
closed_at TEXT,
|
||
close_price REAL,
|
||
close_reason TEXT,
|
||
theoretical_pnl REAL, -- 理论盈亏%（基于中值买入价）
|
||
-- 实操数据（由用户或导入脚本填入）
|
||
actual_action TEXT, -- "买入600股@148.86"
|
||
actual_entry REAL,
|
||
actual_shares INTEGER,
|
||
actual_exit REAL,
|
||
actual_pnl REAL,
|
||
actual_exit_reason TEXT,
|
||
notes TEXT
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_track_code ON strategy_tracking(code);
|
||
CREATE INDEX IF NOT EXISTS idx_track_status ON strategy_tracking(status);
|
||
CREATE INDEX IF NOT EXISTS idx_track_date ON strategy_tracking(tracked_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
|
||
# ── 三值 RR 迁移（2026-07-22 老爸）：买入区下沿/中值/上沿三个 RR ──
|
||
# rr_ratio 保留=中值 RR（排序/门槛沿用），新增 rr_low / rr_high 展示用。
|
||
for _col in ("rr_low REAL DEFAULT 0", "rr_high REAL DEFAULT 0"):
|
||
try:
|
||
conn.execute(f"ALTER TABLE holding_strategies ADD COLUMN {_col}")
|
||
except sqlite3.OperationalError:
|
||
pass
|
||
# ── rec_score 迁移（2026-07-27）：五维复合推荐评分 0-100 ──
|
||
try:
|
||
conn.execute("ALTER TABLE holding_strategies ADD COLUMN rec_score INTEGER DEFAULT 0")
|
||
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 reconcile_signal_from_analysis(conn, code: str) -> str:
|
||
"""以已存 full_analysis 为唯一事实源，重算 timing_signal 并写回。
|
||
根治"信号与分析脱节"（per_stock 分开写信号/分析导致的 信号=买入但分析=观望）。
|
||
返回最终信号。无裁决行 → 清空动作级信号（防陈旧买入残留）。"""
|
||
try:
|
||
row = conn.execute("SELECT full_analysis, timing_signal FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||
if not row:
|
||
return ""
|
||
fa, old_sig = row[0] or "", row[1] or ""
|
||
verdict = ""
|
||
for line in fa.split("\n"):
|
||
if "【综合结论】" in line:
|
||
for s in ("买入", "可买入", "可加仓", "卖出", "止盈", "关注", "观望", "持有", "弱势持有"):
|
||
if s in line:
|
||
verdict = s
|
||
break
|
||
break
|
||
if verdict:
|
||
new_sig = verdict
|
||
# ── 矛盾降级（2026-07-24 老爸：688660综合结论=买入但操作建议说"继续空仓观望暂不执行"）──
|
||
# 【综合结论】给方向，【操作建议】/【建议仓位】给执行。执行层明确否定买入时，
|
||
# 以执行为准——信号降级为关注，tag/exec 一律不许升。
|
||
if new_sig in ("买入", "可买入", "可加仓"):
|
||
_NEG_ADVICE = ("继续空仓", "暂不执行", "不宜买入", "不买入", "不建仓", "不新建仓",
|
||
"等待企稳", "暂缓买入", "保持空仓", "维持空仓", "不建议买入", "空仓观望",
|
||
"不建议操作", "等待价格回落", "等待回调", "高于买入区上沿")
|
||
for line in fa.split("\n"):
|
||
if "【建议仓位】" in line and ("不新建仓" in line or "不建仓" in line):
|
||
new_sig = "关注"
|
||
print(f" [RECONCILE] {code} 建议仓位=不建仓('{line.strip()[:40]}')，信号降级为关注", flush=True)
|
||
break
|
||
if ("【操作建议】" in line or line.strip().startswith("【操作建议】")) \
|
||
and any(k in line for k in _NEG_ADVICE):
|
||
new_sig = "关注"
|
||
print(f" [RECONCILE] {code} 操作建议否定买入('{line.strip()[:40]}')，信号降级为关注", flush=True)
|
||
break
|
||
elif old_sig in ("买入", "可买入", "可加仓"):
|
||
new_sig = "" # 无裁决且陈旧动作信号 → 清除
|
||
else:
|
||
new_sig = old_sig
|
||
if new_sig != old_sig:
|
||
conn.execute("UPDATE holding_strategies SET timing_signal=? WHERE code=? AND status='active'",
|
||
(new_sig, code))
|
||
conn.commit()
|
||
# 信号变了 → tag 跟着对齐
|
||
sync_recommend_tag(conn, code, new_sig)
|
||
print(f" [RECONCILE] {code} 信号 {old_sig}→{new_sig}（以分析为准）", flush=True)
|
||
return new_sig
|
||
except Exception as e:
|
||
print(f" [RECONCILE] {code} 异常: {e}", flush=True)
|
||
return ""
|
||
|
||
|
||
def recompute_rr(conn, code: str) -> float:
|
||
"""用买入区+已存止损/止盈重算三值 RR 并写回（rr_low/rr_ratio中值/rr_high）。
|
||
根治"LLM 不输出 RR → rr_ratio 永远 0"的断链（红线：RR 由系统算，不信 LLM）。
|
||
公式: RR(x) = (上方目标 - x) / (x - 止损)；x 分别取买入区下沿/中值/上沿。
|
||
上方目标基于中值参考价统一定（不因 x 不同而漂移），保证三值自洽：
|
||
rr_low > rr_ratio > rr_high 恒成立，展示"在同一阻力下不同入场价的敏感度"。
|
||
目标 = min(止盈, 20日新高若高于中值)（2026-07-24 老爸：
|
||
前高挡在中间时止盈是放空炮，真实RR必须对最近上方阻力先结算）。
|
||
rr_ratio=中值 RR 用于排序与2.0门槛；rr_low/rr_high 展示入场价敏感度。
|
||
区间缺失 → 中值兜底现价（low/high=0）；损/盈缺失或 x<=止损 → 该值=0。"""
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT entry_low, entry_high, stop_loss, take_profit FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
if not row:
|
||
return 0.0
|
||
el, eh, sl, tp = (row[0] or 0), (row[1] or 0), (row[2] or 0), (row[3] or 0)
|
||
|
||
# 卖出/止盈信号：RR（买在区间的盈亏比）对卖出无意义，直接返回0（2026-07-27 老爸）
|
||
try:
|
||
_sig_r = conn.execute("SELECT timing_signal FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||
if _sig_r and _sig_r[0] in ("卖出", "止盈"):
|
||
conn.execute("UPDATE holding_strategies SET rr_ratio=0, rr_low=0, rr_high=0 WHERE code=? AND status='active'", (code,))
|
||
conn.commit()
|
||
return 0.0
|
||
except Exception:
|
||
pass
|
||
|
||
# 20日新高（前高阻力）
|
||
high_20d = 0.0
|
||
try:
|
||
r20 = conn.execute(
|
||
"SELECT MAX(high) FROM (SELECT high FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 20)",
|
||
(code,)).fetchone()
|
||
if r20 and r20[0]:
|
||
high_20d = float(r20[0])
|
||
except Exception:
|
||
pass
|
||
|
||
# 基准参考价 = 区间中值（决定上方目标，三值共用）
|
||
ref = (el + eh) / 2.0 if el > 0 and eh > el else 0
|
||
target = tp
|
||
# 已持仓股不适用20日新高阻力（用户已按原始推荐买入，RR应保持原值）
|
||
_owned = conn.execute("SELECT shares FROM holdings WHERE code=? AND is_active=1 AND shares>0", (code,)).fetchone()
|
||
if not _owned and ref > 0 and high_20d > ref and high_20d < tp:
|
||
target = high_20d
|
||
|
||
def _rr(x):
|
||
if sl > 0 and target > 0 and x > sl and target > x:
|
||
v = round((target - x) / (x - sl), 2)
|
||
return v if v > 0 else 0.0
|
||
return 0.0
|
||
|
||
rr_low = rr_mid = rr_high = 0.0
|
||
if el > 0 and eh > el:
|
||
rr_low = _rr(el) # 下沿买入：最乐观
|
||
rr_mid = _rr((el + eh) / 2.0)
|
||
rr_high = _rr(eh) # 上沿买入：最保守
|
||
else:
|
||
# 区间缺失 → 中值兜底现价
|
||
try:
|
||
pr = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
|
||
if pr and (pr[0] or 0) > 0:
|
||
rr_mid = _rr(float(pr[0]))
|
||
except Exception:
|
||
pass
|
||
conn.execute(
|
||
"UPDATE holding_strategies SET rr_ratio=?, rr_low=?, rr_high=? WHERE code=? AND status='active'",
|
||
(rr_mid, rr_low, rr_high, code))
|
||
conn.commit()
|
||
compute_rec_score(conn, code) # RR 变→评分同步刷新
|
||
return rr_mid
|
||
except Exception as e:
|
||
print(f" [RR] {code} 重算失败: {e}", flush=True)
|
||
return 0.0
|
||
|
||
|
||
def compute_rec_score(conn, code: str) -> int:
|
||
"""五维复合推荐评分 0-100。RR高≠值得买，趋势+行业+信号综合判断。
|
||
维度：RR(0-35) + 信号(0-25) + 趋势(0-20) + 行业(0-10) + 区间(0-10)"""
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT rr_ratio, timing_signal, tech_snapshot, sector_context, entry_low, entry_high "
|
||
"FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||
if not row:
|
||
return 0
|
||
rr, sig, tech, sector, el, eh = row
|
||
rr = rr or 0; el = el or 0; eh = eh or 0
|
||
|
||
# ── 1. RR (0-35) ──
|
||
if rr >= 3.0: s_rr = 35
|
||
elif rr >= 2.5: s_rr = 28
|
||
elif rr >= 2.0: s_rr = 20
|
||
elif rr >= 1.5: s_rr = 10
|
||
else: s_rr = 0
|
||
|
||
# ── 2. 信号强度 (0-25) ──
|
||
sig_map = {"买入": 25, "可买入": 20, "可加仓": 15}
|
||
s_sig = sig_map.get(sig, 0)
|
||
|
||
# ── 3. 技术趋势 (0-20) ──
|
||
tech_str = str(tech or '')
|
||
# 形态判定
|
||
if '/bullish' in tech_str or '看涨' in tech_str:
|
||
s_trend = 15
|
||
elif '/bearish' in tech_str or '看跌' in tech_str:
|
||
s_trend = 8
|
||
else:
|
||
s_trend = 12
|
||
# MA 排列加成
|
||
import re as _re_ma
|
||
ma_vals = {}
|
||
for m in _re_ma.finditer(r'MA(\d+)=([\d.]+)', tech_str):
|
||
ma_vals[int(m.group(1))] = float(m.group(2))
|
||
if all(k in ma_vals for k in [5,10,20,60]):
|
||
if ma_vals[5] > ma_vals[10] > ma_vals[20] > ma_vals[60]:
|
||
s_trend += 5 # 多头排列
|
||
elif ma_vals[5] < ma_vals[10] < ma_vals[20] < ma_vals[60]:
|
||
s_trend -= 3 # 空头排列
|
||
s_trend = max(0, min(20, s_trend))
|
||
|
||
# ── 4. 行业强弱 (0-10) ──
|
||
sec_str = str(sector or '')
|
||
if '领涨' in sec_str:
|
||
s_sec = 9
|
||
elif '偏强' in sec_str or '上涨' in sec_str:
|
||
s_sec = 7
|
||
elif '偏弱' in sec_str or '下跌' in sec_str:
|
||
s_sec = 3
|
||
else:
|
||
s_sec = 5
|
||
|
||
# ── 5. 买入区间质量 (0-10) ──
|
||
s_zone = 0
|
||
if el > 0 and eh > el:
|
||
zone_pct = (eh - el) / el * 100
|
||
if zone_pct >= 5: s_zone = 10
|
||
elif zone_pct >= 3: s_zone = 7
|
||
elif zone_pct >= 2: s_zone = 4
|
||
else: s_zone = 2
|
||
|
||
total = s_rr + s_sig + s_trend + s_sec + s_zone
|
||
conn.execute(
|
||
"UPDATE holding_strategies SET rec_score=? WHERE code=? AND status='active'",
|
||
(total, code))
|
||
conn.commit()
|
||
# 高评分自动打推荐 tag（补 LLM 未打 tag 的缺口）
|
||
if total >= 50 and rr >= 2.0 and sig in ("买入", "可买入", "可加仓"):
|
||
_pos_v = conn.execute(
|
||
"SELECT position_advice FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
if _pos_v and _pos_v[0] and '%' in str(_pos_v[0]):
|
||
conn.execute(
|
||
"UPDATE holding_strategies SET tag='current_recommend' WHERE code=? AND status='active' AND (tag IS NULL OR tag='')",
|
||
(code,))
|
||
conn.commit()
|
||
return total
|
||
except Exception as e:
|
||
print(f" [SCORE] {code} 评分失败: {e}", flush=True)
|
||
return 0
|
||
|
||
|
||
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）。"""
|
||
try:
|
||
# 2026-08-18 修复：RR 由 scanner 定（老莫原则），已有值不重算覆盖。
|
||
# 原逻辑每次信号同步都 recompute_rr → 重评后参数漂移 → scanner 定的 RR(2.29) 被悄悄改成 1.51。
|
||
_rr_have = conn.execute("SELECT rr_ratio FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||
if not (_rr_have and (_rr_have[0] or 0) > 0):
|
||
recompute_rr(conn, code) # 仅 RR 缺失(0)时兜底，保证 tag/入队/盯盘排序有值
|
||
_row = conn.execute(
|
||
"SELECT tag, timing_signal FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||
_old_tag = (_row[0] or '') if _row else ''
|
||
_old_sig = (_row[1] or '') if _row else ''
|
||
# 卖出/止盈 仅对持仓股算动作信号（没持仓卖什么）
|
||
_ACTION_BUY = ("买入", "可买入", "可加仓")
|
||
_ACTION_SELL = ("卖出", "止盈")
|
||
if timing_signal in _ACTION_SELL:
|
||
_h = conn.execute("SELECT shares FROM holdings WHERE code=? AND is_active=1", (code,)).fetchone()
|
||
if not (_h and (_h[0] or 0) > 0):
|
||
timing_signal = "" # 非持仓的卖出/止盈不算动作信号
|
||
# ── 仓位自动补全（2026-07-27 老爸：不明确就让它明确，不是丢弃）──
|
||
# LLM 经常输出"减仓或观望/中等仓位"等模糊表述，系统按公式自动计算。
|
||
# 基础仓位 by RR(<1.5→不推荐,1.5~3→8%,3~5→12%,5+→15%) × 成长系数0.85(兜底)
|
||
# → 最终范围5-20%
|
||
if timing_signal in _ACTION_BUY:
|
||
import re as _re2
|
||
_pos_r = conn.execute(
|
||
"SELECT rr_ratio, position_advice FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||
if _pos_r and not _re2.search(r'\d+(?:\.\d+)?\s*%', _pos_r[1] or ''):
|
||
_rr_pos = float(_pos_r[0] or 0)
|
||
if _rr_pos < 1.5:
|
||
_pct = 0 # RR不推荐
|
||
elif _rr_pos < 3:
|
||
_pct = 8
|
||
elif _rr_pos < 5:
|
||
_pct = 12
|
||
else:
|
||
_pct = 15
|
||
# 系数兜底：成长股0.85（最保守），大盘系数1.0（中性）
|
||
_pct = round(_pct * 0.85, 0)
|
||
_pct = max(5, min(20, _pct))
|
||
_pos_auto = f"{int(_pct)}%（系统按RR{_rr_pos:.1f}自动计算，见原建议仓位）"
|
||
# ── 股数换算（2026-07-27 老爸：方便快速操作）──
|
||
# 2026-07-27 修正：按总资产算仓位，不用现金（现金波动剧烈）
|
||
try:
|
||
_lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
|
||
_ta = conn.execute("SELECT total_assets FROM portfolio_summary WHERE id=1").fetchone()
|
||
if _lp and _lp[0] and _ta and _ta[0]:
|
||
_price = float(_lp[0])
|
||
_total = float(_ta[0])
|
||
_shares_raw = _total * _pct / 100.0 / _price
|
||
if _shares_raw >= 100:
|
||
_shares = int(_shares_raw / 100) * 100 # A股整手
|
||
else:
|
||
_shares = int(_shares_raw)
|
||
if _shares > 0:
|
||
_pos_auto = f"{int(_pct)}% ≈ {_shares}股（系统按RR{_rr_pos:.1f}自动计算）"
|
||
except Exception:
|
||
pass
|
||
conn.execute(
|
||
"UPDATE holding_strategies SET position_advice=? WHERE code=? AND status='active'",
|
||
(_pos_auto, code))
|
||
conn.commit()
|
||
print(f" [AUTO-POS] {code} 仓位'{_pos_r[1]}'→'{_pos_auto}'", flush=True)
|
||
if timing_signal in _ACTION_BUY or timing_signal in _ACTION_SELL:
|
||
# ── 买入RR质量门禁（2026-07-27 老爸：RR<2.0的买入不值得推荐，tag都不能打，防盯盘垃圾）──
|
||
# 此前tag先打、enqueue再查RR，结果RR<2.0的tag已落盯盘，与XMPP不一致。
|
||
# 已持仓股不再重复推荐（2026-07-28 老爸：已买了的票该在持仓不在推荐）
|
||
_should_tag = True
|
||
_owned = conn.execute(
|
||
"SELECT shares FROM holdings WHERE code=? AND is_active=1 AND shares>0",
|
||
(code,)).fetchone()
|
||
if _owned and _owned[0] and _owned[0] > 0:
|
||
_should_tag = False # 已持仓，不再重复推荐
|
||
if _old_tag == 'current_recommend':
|
||
conn.execute("UPDATE holding_strategies SET tag='' WHERE code=? AND status='active'", (code,))
|
||
conn.commit()
|
||
elif timing_signal in _ACTION_BUY:
|
||
_rr_chk = conn.execute(
|
||
"SELECT rr_ratio FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||
if _rr_chk and (_rr_chk[0] or 0) < 2.0:
|
||
print(f" [TAG SYNC] {code} RR={_rr_chk[0]:.2f}<2.0，不打推荐tag", flush=True)
|
||
_should_tag = False
|
||
if _should_tag:
|
||
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()
|
||
else:
|
||
# RR不达标 → 清除旧tag（如果打了）
|
||
if _old_tag == 'current_recommend':
|
||
conn.execute("UPDATE holding_strategies SET tag='' WHERE code=? AND status='active'", (code,))
|
||
conn.commit()
|
||
# 入队条件（2026-07-27 老爸：tag新鲜转成、信号转入动作级、或卖出信号 — 三个场景均需入队）
|
||
_old_action = _old_sig in _ACTION_BUY or _old_sig in _ACTION_SELL
|
||
_new_action = timing_signal in _ACTION_BUY or timing_signal in _ACTION_SELL
|
||
if (_should_tag and _old_tag != 'current_recommend') or (not _old_action and _new_action):
|
||
enqueue_recommend(conn, code) # 新推荐 → 摘要队列（batch 结束统一发）
|
||
elif _old_tag == 'current_recommend':
|
||
# 空信号或非动作信号 → 清除自动推荐（active_manual 不动）
|
||
conn.execute(
|
||
"UPDATE holding_strategies SET tag='' "
|
||
"WHERE code=? AND status='active' AND tag='current_recommend'",
|
||
(code,))
|
||
conn.commit()
|
||
# ── 策略版本追踪（2026-07-27 老爸：每次tag变更都记录到评估表）──
|
||
track_strategy_version(conn, code)
|
||
except Exception as e:
|
||
print(f" [TAG SYNC] {code} 失败: {e}", flush=True)
|
||
|
||
|
||
def enqueue_recommend(conn, code: str):
|
||
"""新推荐入摘要队列（防逐只轰炸）。batch_reassess 跑完后 flush_rec_digest 统一发一条。
|
||
校验（2026-07-24 老爸"阿猫阿狗"事件后加严）：
|
||
1. tag=current_recommend 且信号为动作级
|
||
2. RR(中值)>=2.0（1.5边缘的平庸推荐一律拦下）
|
||
3. position_advice 必须含明确仓位%（"减仓或观望/不新建仓"不算推荐）
|
||
4. 买入区必须有效（区—~—/0~0 不入）
|
||
5. 买入信号时现价不得在区上沿 5% 以上（追高不买）"""
|
||
try:
|
||
import json as _j, re as _re
|
||
from datetime import datetime as _dt
|
||
row = conn.execute(
|
||
"SELECT name, timing_signal, tag, entry_low, entry_high, stop_loss, take_profit, "
|
||
"rr_ratio, rr_low, rr_high, position_advice, full_analysis, rec_score FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
if not row:
|
||
return
|
||
name, sig, tag, el, eh, sl, tp, rr, rr_lo, rr_hi, pos, fa, score = row
|
||
if tag != 'current_recommend' or sig not in ("买入", "可买入", "可加仓", "卖出", "止盈"):
|
||
print(f" [REC] {code} 非有效推荐(tag={tag},sig={sig})，不入队", flush=True)
|
||
return False
|
||
# ── 买入类质量闸（卖出/止盈不受 RR/仓位限制——那是风控动作）──
|
||
if sig in ("买入", "可买入", "可加仓"):
|
||
if (rr or 0) < 2.0:
|
||
print(f" [REC] {code} RR={rr}<2.0 平庸推荐，不入队", flush=True)
|
||
return False
|
||
if not _re.search(r'\d+(?:\.\d+)?\s*%', pos or ''):
|
||
print(f" [REC] {code} 仓位非明确%({pos})，不入队", flush=True)
|
||
return False
|
||
if not (el and eh and el > 0 and eh > el):
|
||
print(f" [REC] {code} 买入区缺失/无效({el}~{eh})，不入队", flush=True)
|
||
return False
|
||
try:
|
||
_lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
|
||
if _lp and _lp[0] and _lp[0] > eh * 1.05:
|
||
print(f" [REC] {code} 现价{_lp[0]}超区上沿{eh}5%，追高不入队", flush=True)
|
||
return False
|
||
except Exception:
|
||
pass
|
||
# 提取【最终新策略】段作为推荐依据摘要
|
||
fa_text = fa or ""
|
||
strat = ""
|
||
for marker in ("【最终新策略】", "【综合结论】"):
|
||
idx = fa_text.find(marker)
|
||
if idx >= 0:
|
||
strat = fa_text[idx:idx + 450]
|
||
break
|
||
qf = '/home/hmo/MoFin/gateway/logs/rec_digest_queue.jsonl'
|
||
import os as _os
|
||
_os.makedirs(_os.path.dirname(qf), exist_ok=True)
|
||
with open(qf, 'a', encoding='utf-8') as f:
|
||
f.write(_j.dumps({"code": code, "name": name, "signal": sig,
|
||
"entry_low": el, "entry_high": eh, "stop_loss": sl,
|
||
"take_profit": tp, "rr": rr, "rr_low": rr_lo, "rr_high": rr_hi,
|
||
"position": pos, "score": score or 0,
|
||
"strategy_excerpt": strat,
|
||
"full_analysis": fa_text[:2500],
|
||
"ts": _dt.now().isoformat()}, ensure_ascii=False) + "\n")
|
||
print(f" [REC] {code} 已入推荐摘要队列", flush=True)
|
||
return True
|
||
except Exception as e:
|
||
print(f" [REC] {code} 入队失败: {e}", flush=True)
|
||
return False
|
||
|
||
|
||
def track_strategy_version(conn, code: str):
|
||
"""版本化策略追踪：每次策略变更自动记录新版本。
|
||
跟踪所有策略状态（不限于 tag='current_recommend'），tag 清除时也记录。"""
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT name, timing_signal, rec_score, rr_ratio, entry_low, entry_high, "
|
||
"stop_loss, take_profit, position_advice, tag FROM holding_strategies "
|
||
"WHERE code=? AND status='active'", (code,)).fetchone()
|
||
if not row:
|
||
return
|
||
name, sig, score, rr, el, eh, sl, tp, pos, tag = row
|
||
# 空壳策略（无信号/无买入区）不追踪
|
||
if not sig or (not el and not eh):
|
||
return
|
||
|
||
lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
|
||
price = lp[0] if lp and lp[0] else 0
|
||
mid = round((el + eh) / 2, 2) if el > 0 and eh > el else 0
|
||
|
||
# 查上一个版本
|
||
prev = conn.execute(
|
||
"SELECT entry_low, entry_high, stop_loss, take_profit, rr_ratio, rec_score, "
|
||
"status FROM strategy_tracking WHERE code=? ORDER BY id DESC LIMIT 1",
|
||
(code,)).fetchone()
|
||
|
||
# 计算版本号
|
||
# 计算版本号
|
||
last_ver = conn.execute(
|
||
"SELECT MAX(version_seq) FROM strategy_tracking WHERE code=?", (code,)).fetchone()
|
||
ver = (last_ver[0] or 0) + 1 if last_ver else 1
|
||
|
||
if prev and prev[6] == 'active': # 上一版本还在进行中
|
||
if (abs((prev[0] or 0) - (el or 0)) < 0.01 and
|
||
abs((prev[1] or 0) - (eh or 0)) < 0.01 and
|
||
abs((prev[2] or 0) - (sl or 0)) < 0.01 and
|
||
abs((prev[3] or 0) - (tp or 0)) < 0.01):
|
||
# 参数没变，只更新评分和RR
|
||
conn.execute(
|
||
"UPDATE strategy_tracking SET rec_score=?, rr_ratio=?, price_at_track=?, "
|
||
"timing_signal=?, position_advice=? WHERE id=("
|
||
"SELECT id FROM strategy_tracking WHERE code=? ORDER BY id DESC LIMIT 1)",
|
||
(score, rr, price, sig, pos, code))
|
||
conn.commit()
|
||
return
|
||
|
||
# 有变更 → 追加新版本
|
||
change = ""
|
||
if prev:
|
||
parts = []
|
||
if abs((prev[0] or 0) - (el or 0)) > 0.5: parts.append(f"区{prev[0]}→{el}")
|
||
if abs((prev[2] or 0) - (sl or 0)) > 0.5: parts.append(f"损{prev[2]}→{sl}")
|
||
if abs((prev[3] or 0) - (tp or 0)) > 0.5: parts.append(f"盈{prev[3]}→{tp}")
|
||
if abs((prev[5] or 0) - (score or 0)) >= 5: parts.append(f"评分{prev[5]}→{score}")
|
||
change = "; ".join(parts) if parts else ""
|
||
|
||
conn.execute("""
|
||
INSERT INTO strategy_tracking
|
||
(code, name, version_seq, tracked_at, timing_signal, rec_score, rr_ratio,
|
||
entry_low, entry_high, entry_mid, stop_loss, take_profit,
|
||
position_advice, price_at_track, change_summary)
|
||
VALUES (?,?,?,datetime('now','localtime'),?,?,?,?,?,?,?,?,?,?,?)
|
||
""", (code, name, ver, sig, score, rr, el, eh, mid, sl, tp, pos, price, change))
|
||
conn.commit()
|
||
if change:
|
||
print(f" [TRACK] {code} v{ver}: {change}", flush=True)
|
||
except Exception as e:
|
||
print(f" [TRACK] {code} 版本记录失败: {e}", flush=True)
|
||
|
||
|
||
def check_strategy_outcomes(conn):
|
||
"""检查所有 active 追踪版本是否触发 SL/TP，自动关闭并记录理论盈亏"""
|
||
active = conn.execute("""
|
||
SELECT id, code, entry_mid, stop_loss, take_profit, rr_ratio
|
||
FROM strategy_tracking WHERE status='active'
|
||
""").fetchall()
|
||
|
||
updated = 0
|
||
for r in active:
|
||
tid, code, mid, sl, tp, rr = r
|
||
lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
|
||
if not lp or not lp[0]:
|
||
continue
|
||
price = float(lp[0])
|
||
mid = mid or price # fallback
|
||
|
||
closed = False
|
||
# ── 持仓成本感知（2026-07-29 老爸抓包：策略版"止盈"可能远低于真实成本，必须区分语义）──
|
||
_hc = conn.execute("SELECT cost, shares FROM holdings WHERE code=? AND is_active=1 AND shares>0", (code,)).fetchone()
|
||
_cost = float(_hc[0]) if _hc and _hc[0] else 0.0
|
||
_real_pnl = round((price - _cost) / _cost * 100, 1) if _cost > 0 else None
|
||
if tp and tp > 0 and price >= tp:
|
||
pnl_pct = round((tp - mid) / mid * 100, 1) if mid > 0 else 0
|
||
_reason = '止盈触发' if (_real_pnl is None or _real_pnl > 0) else '反弹减仓触发'
|
||
conn.execute("""
|
||
UPDATE strategy_tracking SET status='hit_tp', closed_at=datetime('now','localtime'),
|
||
close_price=?, close_reason=?, theoretical_pnl=?
|
||
WHERE id=?
|
||
""", (price, _reason, pnl_pct, tid))
|
||
if _real_pnl is not None and _real_pnl <= 0:
|
||
print(f" [TRACK] {code} v{tid} 反弹减仓位! {price}≥{tp}，成本{_cost}，浮亏{_real_pnl}%（减仓窗口，非止盈）", flush=True)
|
||
elif _real_pnl is not None and _real_pnl < 2:
|
||
print(f" [TRACK] {code} v{tid} 成本线/阻力位! {price}≥{tp}，成本{_cost}（贴近成本+前高阻力，决策点非止盈）", flush=True)
|
||
elif _real_pnl is not None:
|
||
print(f" [TRACK] {code} v{tid} 止盈! {price}≥{tp} 实际盈利+{_real_pnl}%（成本{_cost}）", flush=True)
|
||
else:
|
||
print(f" [TRACK] {code} v{tid} 追踪版本止盈 {price}≥{tp} +{pnl_pct}%（未持仓）", flush=True)
|
||
closed = True
|
||
elif sl and sl > 0 and price <= sl:
|
||
pnl_pct = round((sl - mid) / mid * 100, 1) if mid > 0 else -5
|
||
conn.execute("""
|
||
UPDATE strategy_tracking SET status='hit_sl', closed_at=datetime('now','localtime'),
|
||
close_price=?, close_reason='止损触发', theoretical_pnl=?
|
||
WHERE id=?
|
||
""", (price, pnl_pct, tid))
|
||
if _real_pnl is not None:
|
||
print(f" [TRACK] {code} v{tid} 止损! {price}≤{sl}，成本{_cost}，实际{_real_pnl}%", flush=True)
|
||
else:
|
||
print(f" [TRACK] {code} v{tid} 止损! {price}≤{sl} {pnl_pct}%", flush=True)
|
||
closed = True
|
||
|
||
if closed:
|
||
updated += 1
|
||
|
||
if updated:
|
||
conn.commit()
|
||
# ── 统计验证（2026-07-28 老爸：胜率/夏普/最大回撤）──
|
||
try:
|
||
closed = conn.execute("""
|
||
SELECT theoretical_pnl FROM strategy_tracking
|
||
WHERE status IN ("hit_tp", "hit_sl", "manual_close") AND theoretical_pnl IS NOT NULL
|
||
""").fetchall()
|
||
if closed:
|
||
wins = [p[0] for p in closed if p[0] > 0]
|
||
losses = [abs(p[0]) for p in closed if p[0] < 0]
|
||
win_rate = len(wins) / len(closed) if closed else 0.55
|
||
avg_win = sum(wins) / len(wins) if wins else 0
|
||
avg_loss = sum(losses) / len(losses) if losses else 0
|
||
sharpe = (avg_win * win_rate - avg_loss * (1 - win_rate)) / (avg_loss if avg_loss > 0 else 1) if avg_loss > 0 else 0
|
||
max_dd = max([abs(p[0]) for p in closed if p[0] < 0], default=0)
|
||
print(f" [STATS] 胜率{win_rate:.0%} 夏普{sharpe:.2f} 最大回撤{max_dd:.1f}%", flush=True)
|
||
except Exception as _se:
|
||
print(f" [STATS] 统计失败: {_se}", flush=True)
|
||
return updated
|
||
|
||
|
||
def flush_rec_digest(max_items=5):
|
||
"""把队列里的新推荐聚成一条 XMPP 摘要（RR 降序，最多 max_items 只）。
|
||
头部 1-2 只附带策略依据摘要+按现金的操盘建议。"""
|
||
import json as _j, os as _os, sqlite3 as _sq
|
||
qf = '/home/hmo/MoFin/gateway/logs/rec_digest_queue.jsonl'
|
||
if not _os.path.exists(qf):
|
||
return 0
|
||
try:
|
||
with open(qf, encoding='utf-8') as f:
|
||
items = [_j.loads(l) for l in f if l.strip()]
|
||
except Exception:
|
||
return 0
|
||
if not items:
|
||
return 0
|
||
_os.remove(qf)
|
||
|
||
# ── 快照回库校验（2026-07-24 老爸：推荐和XMPP同步）──
|
||
# 队列是打标瞬间的快照；flush 前回库读实时 信号/RR/tag，
|
||
# 信号降级为弱信号或RR跌破2.0的条目直接丢弃——XMPP说的必须和盯盘一致。
|
||
import sqlite3 as _sq0
|
||
from datetime import datetime as _ddt
|
||
_now = _ddt.now()
|
||
_h, _m, _w = _now.hour, _now.minute, _now.weekday()
|
||
_market_open = _w < 5 and ((_h == 9 and _m >= 30) or (10 <= _h < 15))
|
||
_vconn = _sq0.connect("/home/hmo/MoFin/data/mofin.db")
|
||
_WEAK = ("信号不充分", "关注", "弱势持有", "观望", "持有", "")
|
||
_live = []
|
||
for it in items:
|
||
r = _vconn.execute(
|
||
"SELECT timing_signal, rr_ratio, tag, reassessed_at FROM holding_strategies WHERE code=? AND status='active'",
|
||
(it['code'],)).fetchone()
|
||
if not r:
|
||
print(f" [REC] {it['code']} 已不在库，丢弃", flush=True)
|
||
continue
|
||
cur_sig, cur_rr, cur_tag, cur_ra = r[0] or "", r[1] or 0, r[2] or "", r[3] or ""
|
||
if cur_tag != 'current_recommend':
|
||
print(f" [REC] {it['code']} tag已撤销({cur_tag})，丢弃", flush=True)
|
||
continue
|
||
# ── 数据时效校验（2026-07-27 老爸：盘前分析盘中推送=过期数据误导）──
|
||
# 分析时间在开盘前且现在已开盘 → 触发盘中重评（用实时数据分析，不推旧分析）
|
||
if _market_open and cur_ra:
|
||
try:
|
||
_ra_dt = _ddt.fromisoformat(str(cur_ra)[:19])
|
||
# 分析在 08:00-09:29 之间做的 = 盘前分析 → 盘中触发重评
|
||
if (_ra_dt.hour >= 8 and (_ra_dt.hour < 9 or (_ra_dt.hour == 9 and _ra_dt.minute < 30))) \
|
||
and (_ddt.now() - _ra_dt).total_seconds() > 120:
|
||
try:
|
||
import subprocess as _sp
|
||
_re = _sp.run(
|
||
["python3", "/home/hmo/MoFin/deploy/profile-scripts/per_stock_reassess.py", it['code']],
|
||
capture_output=True, text=True, timeout=90)
|
||
if _re.returncode == 0:
|
||
print(f" [REC] {it['code']} 盘中重评完成，用实时数据更新策略", flush=True)
|
||
# 重新读库获取更新后数据
|
||
r = _vconn.execute(
|
||
"SELECT timing_signal, rr_ratio, tag FROM holding_strategies WHERE code=? AND status='active'",
|
||
(it['code'],)).fetchone()
|
||
if r:
|
||
cur_sig, cur_rr, cur_tag = r[0] or "", r[1] or 0, r[2] or ""
|
||
if cur_tag != 'current_recommend':
|
||
print(f" [REC] {it['code']} 重评后tag撤销，丢弃", flush=True)
|
||
continue
|
||
else:
|
||
print(f" [REC] {it['code']} 盘中重评失败(rc={_re.returncode})，标记警告", flush=True)
|
||
it['_stale_warn'] = "⚠️ 盘前分析（已开盘未能及时重评，请结合实时盘面判断）"
|
||
except subprocess.TimeoutExpired:
|
||
print(f" [REC] {it['code']} 盘中重评超时，标记警告", flush=True)
|
||
it['_stale_warn'] = "⚠️ 盘前分析（已开盘未能及时重评，请结合实时盘面判断）"
|
||
except Exception:
|
||
pass
|
||
if it.get('signal') in ("买入", "可买入", "可加仓"):
|
||
if cur_sig in _WEAK:
|
||
print(f" [REC] {it['code']} 信号降级为'{cur_sig}'，丢弃", flush=True)
|
||
continue
|
||
if cur_rr < 2.0:
|
||
print(f" [REC] {it['code']} 实时RR={cur_rr}<2.0，丢弃", flush=True)
|
||
continue
|
||
it['signal'] = cur_sig # 用实时信号发
|
||
it['rr'] = cur_rr
|
||
_live.append(it)
|
||
_vconn.close()
|
||
items = _live
|
||
if not items:
|
||
print(" [REC] 快照校验后无有效推荐，不发digest", flush=True)
|
||
return 0
|
||
|
||
_SELL_SIGS = ("卖出", "止盈")
|
||
# 卖出/止盈是释放现金的操作，不占买入预算，单独一组排最前
|
||
sells = [x for x in items if x.get('signal') in _SELL_SIGS]
|
||
buys_all = [x for x in items if x.get('signal') not in _SELL_SIGS]
|
||
buys_all.sort(key=lambda x: (x.get('score') or 0, x.get('rr') or 0), reverse=True)
|
||
items = sells + buys_all
|
||
top = items[:max_items]
|
||
|
||
# ── 现金预算（决定操盘建议 + 换仓策略）：只对买入项计算，卖出不占预算 ──
|
||
cash_note = ""
|
||
rotation_note = ""
|
||
buys = []
|
||
queued = []
|
||
try:
|
||
conn = _sq.connect("/home/hmo/MoFin/data/mofin.db")
|
||
conn.row_factory = _sq.Row
|
||
r = conn.execute("SELECT cash, total_assets FROM portfolio_summary WHERE id=1").fetchone()
|
||
if r and r[1]:
|
||
cash, total = r[0] or 0, r[1]
|
||
budget_pct = cash / total * 100
|
||
cum = 0.0
|
||
buys = []
|
||
queued = []
|
||
for it in buys_all: # 只遍历买入项；sells 永远可执行
|
||
import re as _re
|
||
m = _re.search(r'(\d+(?:\.\d+)?)\s*%', it.get('position') or '')
|
||
pct = float(m.group(1)) if m else 8.0
|
||
if (it.get('rr') or 0) >= 2.0 and cum + pct <= budget_pct + 1e-9:
|
||
buys.append((it, pct))
|
||
cum += pct
|
||
else:
|
||
queued.append((it, pct))
|
||
cash_note = (f"现金{cash/10000:.1f}万({budget_pct:.1f}%)｜按预算本次可执行: "
|
||
+ ("、".join(f"{b[0].get('name') or b[0]['code']}≈{b[1]:.0f}%" for b in buys) if buys else "无")
|
||
+ (f"（合计≈{cum:.0f}%）" if buys else ""))
|
||
# ── 换仓策略：有排队推荐时，找可减的弱持仓来腾挪 ──
|
||
if queued:
|
||
weak = conn.execute("""
|
||
SELECT hs.code, hs.name, hs.timing_signal, h.position_pct, h.cost, lp.price, lp.change_pct
|
||
FROM holding_strategies hs
|
||
JOIN holdings h ON hs.code = h.code AND h.is_active = 1
|
||
LEFT JOIN live_prices lp ON hs.code = lp.code
|
||
WHERE hs.status='active' AND h.shares > 0
|
||
AND hs.timing_signal IN ('弱势持有','观望','持有')
|
||
""").fetchall()
|
||
# ── v7.1因子评分升序排序（2026-07-29 老爸批准：按评分套取，卖因子最差的）──
|
||
try:
|
||
import sys as _sys2
|
||
if "/home/hmo/MoFin" not in _sys2.path:
|
||
_sys2.path.insert(0, "/home/hmo/MoFin")
|
||
from backtest_framework import prepare_bars as _pb, compute_single_score as _cs
|
||
from datetime import datetime as _dt2, timedelta as _td2
|
||
_end2 = _dt2.now().strftime('%Y-%m-%d')
|
||
_start2 = (_dt2.now() - _td2(days=150)).strftime('%Y-%m-%d')
|
||
_scored = []
|
||
for w in weak:
|
||
_sc = 0
|
||
try:
|
||
_bars = _pb(w['code'], _start2, _end2)
|
||
if _bars and len(_bars) >= 25:
|
||
_r = _cs(_bars)
|
||
_sc = _r[0] if _r else 0
|
||
except Exception:
|
||
pass
|
||
_scored.append((_sc, w))
|
||
_scored.sort(key=lambda x: x[0]) # 评分最低 = 优先套取
|
||
weak = [w for _, w in _scored]
|
||
print(" [换仓] 因子评分排序: " + ", ".join(f"{w['name']}({s})" for s, w in _scored[:5]), flush=True)
|
||
except Exception as _se:
|
||
print(f" [换仓] 评分排序失败(回退信号排序): {_se}", flush=True)
|
||
weak = sorted(weak, key=lambda w: ({'弱势持有': 0, '观望': 1}.get(w['timing_signal'], 2),
|
||
-(w['position_pct'] or 0)))
|
||
if weak:
|
||
need_pct = queued[0][1]
|
||
plan = []
|
||
freed = 0.0
|
||
for w in weak:
|
||
if freed >= need_pct:
|
||
break
|
||
plan.append(w)
|
||
freed += w["position_pct"] or 0
|
||
q0 = queued[0][0]
|
||
_names = "+".join(str(w['name']) for w in plan)
|
||
_sigs = ",".join(sorted({w['timing_signal'] for w in plan}))
|
||
rotation_note = ("🔄 换仓建议：现金不足买 " + str(q0.get('name') or q0['code'])
|
||
+ f"（RR={q0.get('rr') or 0}）→ 可减 {_names}"
|
||
+ f"（{_sigs}，腾出≈{freed:.0f}%仓位）换入")
|
||
conn.close()
|
||
except Exception as _re:
|
||
print(f" [REC] 换仓计算异常: {_re}", flush=True)
|
||
|
||
lines = [f"📈 新增推荐 {len(items)} 只（按RR排序）:"]
|
||
# 与盯盘推荐区一致的 可执行/排队 徽章（2026-07-24 老爸：推荐和XMPP同步）
|
||
_exec_codes = {b[0]['code'] for b in buys} | {s['code'] for s in sells}
|
||
for i, it in enumerate(top):
|
||
_rr_mid = it.get('rr') or 0
|
||
_rr_lo, _rr_hi = it.get('rr_low') or 0, it.get('rr_high') or 0
|
||
if _rr_lo and _rr_hi and _rr_lo != _rr_hi:
|
||
_lo_val = min(_rr_lo, _rr_hi)
|
||
_hi_val = max(_rr_lo, _rr_hi)
|
||
_rr_txt = f"RR={_rr_mid}({_lo_val}~{_hi_val})"
|
||
else:
|
||
_rr_txt = f"RR={_rr_mid}"
|
||
_el = it.get('entry_low') or 0
|
||
_eh = it.get('entry_high') or 0
|
||
_mid = f"{(_el+_eh)/2:.2f}" if _el > 0 and _eh > _el else "—"
|
||
_badge = "💰可执行" if it['code'] in _exec_codes else "⏳排队"
|
||
_score = it.get('score') or 0
|
||
_score_txt = f" [{_score}分]" if _score else ""
|
||
lines.append(f"• {_badge}{_score_txt} {it.get('name') or it['code']}({it['code']}) {it['signal']}"
|
||
f" 区{_el or '—'}→{_mid}←{_eh or '—'}"
|
||
f" 损{it.get('stop_loss') or '—'} 盈{it.get('take_profit') or '—'}"
|
||
f" {_rr_txt} 仓位{it.get('position') or '—'}")
|
||
if it.get('_stale_warn'):
|
||
lines.append(f" ⚠️ {it['_stale_warn']}")
|
||
# 所有推荐都附完整策略依据
|
||
if it.get('strategy_excerpt'):
|
||
lines.append(f" 依据: {it['strategy_excerpt']}")
|
||
elif it.get('full_analysis'):
|
||
lines.append(f" 依据: {it['full_analysis']}")
|
||
if len(items) > max_items:
|
||
lines.append(f"…另有 {len(items) - max_items} 只详见盯盘推荐操作区")
|
||
if cash_note:
|
||
lines.append("💰 " + cash_note)
|
||
if rotation_note:
|
||
lines.append(rotation_note)
|
||
try:
|
||
import sys as _s, os as _o2
|
||
_s.path.insert(0, '/home/hmo/MoFin/deploy/profile-scripts')
|
||
from alert_helper import notify, ACTION
|
||
return notify("推荐操作", "\n".join(lines), ACTION)
|
||
except Exception as e:
|
||
print(f" [REC] 摘要推送失败: {e}", flush=True)
|
||
return False
|
||
|
||
|
||
def push_recommend_alert(conn, code: str):
|
||
"""推荐操作 XMPP 推送（tag 转为 current_recommend 时调用，全路径统一）。
|
||
质量门禁：实时价>0、区间有效(下沿<上沿<下沿x3)、现价不超上沿5%、
|
||
损<下沿且在(0.5x~1.0x)现价内、盈>上沿>损。不过不推。"""
|
||
try:
|
||
row = conn.execute(
|
||
"SELECT name, timing_signal, entry_low, entry_high, stop_loss, take_profit, "
|
||
"rr_ratio, position_advice FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
if not row:
|
||
return False
|
||
name, sig, el, eh, sl, tp, rr, pos = row
|
||
lp = conn.execute("SELECT price FROM live_prices WHERE code=?", (code,)).fetchone()
|
||
price = lp[0] if lp and lp[0] else 0
|
||
el, eh, sl, tp = el or 0, eh or 0, sl or 0, tp or 0
|
||
# ── 门禁 ──
|
||
if price <= 0:
|
||
print(f" [ALERT] {code} 无实时价，不推", flush=True); return False
|
||
if not (el > 0 and eh > el and eh < el * 3):
|
||
print(f" [ALERT] {code} 区间无效({el}~{eh})，不推", flush=True); return False
|
||
if price > eh * 1.05:
|
||
print(f" [ALERT] {code} 现价{price}高超上沿{eh}5%，不推", flush=True); return False
|
||
if not (sl > 0 and sl < el and price * 0.5 <= sl <= price):
|
||
print(f" [ALERT] {code} 止损{sl}不合理，不推", flush=True); return False
|
||
if not (tp > eh and tp > sl):
|
||
print(f" [ALERT] {code} 止盈{tp}不合理，不推", flush=True); return False
|
||
import sys as _s, os as _o
|
||
_s.path.insert(0, _o.path.dirname(_o.path.abspath(__file__)))
|
||
from alert_helper import notify, ACTION
|
||
_mid_xmpp = f"{(el+eh)/2:.2f}" if el > 0 and eh > el else "—"
|
||
msg = (f"📈 {name or code}({code}) 价{price}→12维{sig}！"
|
||
f"区间{el}→{_mid_xmpp}←{eh} 损{sl} 盈{tp} RR={rr or 0} 仓位{pos or '-'}")
|
||
return notify("买入信号", msg, ACTION)
|
||
except Exception as e:
|
||
print(f" [ALERT] {code} 推送异常: {e}", flush=True)
|
||
return False
|
||
|
||
|
||
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, strategy_name, 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, strategy_name, 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],
|
||
row[14], source_trigger, row[15], 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 = ("买入", "可买入", "可加仓", "卖出", "止盈")
|
||
_RISK_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 = ''
|
||
_old_sig = ''
|
||
_old_ra = ''
|
||
_old_action = ''
|
||
if True:
|
||
try:
|
||
_old = conn.execute("SELECT full_analysis, reassessed_at, tag, timing_signal, action 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 ''
|
||
_old_sig = _old[3] or ''
|
||
_old_ra = _old[1] or ''
|
||
_old_action = _old[4] or ''
|
||
except:
|
||
pass
|
||
# ── 信号权威层级（2026-07-22）：新鲜(<20h)12维动作级信号，
|
||
# 技术路径(regenerate_all/price_monitor)无权降级为 关注/信号不充分/持有。
|
||
# 只有 LLM 路径(batch_12d/per_stock_12d)可以覆盖。
|
||
# 2026-07-27 老爸补：卖出/止盈的保护漏了（技术路径把卖出→持有导致盯盘显示矛盾）──
|
||
_TECHNICAL_PATHS = ('write_holding_strategy',)
|
||
if source_trigger in _TECHNICAL_PATHS \
|
||
and _old_sig in ("买入", "可买入", "可加仓", "卖出", "止盈") \
|
||
and _new_sig not in _ACTION_SIGNALS and _old_ra:
|
||
try:
|
||
from datetime import datetime as _ddt, timedelta as _dtd
|
||
_ra_dt = _ddt.fromisoformat(str(_old_ra)[:19])
|
||
if (_ddt.now() - _ra_dt) < _dtd(hours=20):
|
||
print(f" [AUTHORITY] {code} 保留新鲜12维信号'{_old_sig}'({_old_ra[:16]})，"
|
||
f"拒绝技术路径降级为'{_new_sig}'", flush=True)
|
||
_new_sig = _old_sig
|
||
data['timing_signal'] = _old_sig
|
||
except Exception:
|
||
pass
|
||
# ── 策略参数权威保护（2026-07-27 老爸：技术路径每2分钟覆写12维的Zone/SL/TP/Position→RR波动→盯盘和XMPP不一致）──
|
||
# 新鲜(<20h)12维分析的技术参数+仓位不允许被技术路径覆写。
|
||
# 2026-07-27 坑：_old_ra=None 时权威保护永不触发（很多股票的reassessed_at为空），
|
||
# 导致系统自动计算的仓位被反复踩回"中等仓位"。加入兜底：仓位含"%（系统按"即永保。
|
||
if source_trigger in _TECHNICAL_PATHS:
|
||
# 兜底：系统自动计算的仓位永久保护（不含 %（系统按 的不保护，即只有 LLM 仓和系统仓被保护）
|
||
_old_pos = conn.execute(
|
||
"SELECT position_advice FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
_old_pos_val = (_old_pos[0] or '') if _old_pos else ''
|
||
if _old_pos_val and '系统按' in str(_old_pos_val):
|
||
data['position_advice'] = _old_pos_val
|
||
# 被保护仓位触发时顺便保护参数（无论 _old_ra 是否空）
|
||
_op = conn.execute(
|
||
"SELECT entry_low, entry_high, stop_loss, take_profit FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
if _op and float(_op[0] or 0) > 0:
|
||
data['entry_low'] = float(_op[0])
|
||
data['entry_high'] = float(_op[1])
|
||
data['stop_loss'] = float(_op[2])
|
||
data['take_profit'] = float(_op[3])
|
||
print(f" [AUTHORITY-POS] {code} 保护系统仓位'{_old_pos_val[:30]}'", flush=True)
|
||
elif _old_ra:
|
||
try:
|
||
from datetime import datetime as _ddt3, timedelta as _dtd3
|
||
_ra_dt3 = _ddt3.fromisoformat(str(_old_ra)[:19])
|
||
if (_ddt3.now() - _ra_dt3) < _dtd3(hours=20):
|
||
_old_params = conn.execute(
|
||
"SELECT entry_low, entry_high, stop_loss, take_profit, position_advice FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
if _old_params:
|
||
_keys = ['entry_low','entry_high','stop_loss','take_profit','position_advice']
|
||
_vals = [v if v else '' for v in _old_params]
|
||
for i, k in enumerate(_keys):
|
||
if i < 4 and float(_vals[i] or 0) > 0:
|
||
data[k] = float(_vals[i])
|
||
elif i == 4 and str(_vals[i]).strip():
|
||
data[k] = str(_vals[i])
|
||
print(f" [AUTHORITY-PARAM] {code} 保留12维参数(区{_vals[0]}~{_vals[1]} 损{_vals[2]} 盈{_vals[3]} pos={_vals[4]})", flush=True)
|
||
except Exception:
|
||
pass
|
||
# ── action 权限保护（与信号同一权威层级，2026-07-22）──
|
||
# 技术路径不得覆盖新鲜(<20h)12维 action。
|
||
# 根治：技术路径写的"盈亏比不足1:1.5不建议买入"旧 action 与12维买入分析同框矛盾。
|
||
if source_trigger not in ('batch_12d', 'per_stock_12d') and _old_action and _old_ra:
|
||
try:
|
||
from datetime import datetime as _ddt2, timedelta as _dtd2
|
||
if (_ddt2.now() - _ddt2.fromisoformat(str(_old_ra)[:19])) < _dtd2(hours=20):
|
||
data['action'] = _old_action
|
||
except Exception:
|
||
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 and source_trigger in ('batch_12d', 'per_stock_12d'):
|
||
_existing_tag = 'current_recommend' # 仅 LLM 路径可创建推荐（防技术路径抖动）
|
||
elif _new_sig and _old_tag == 'current_recommend' and source_trigger in ('batch_12d', 'per_stock_12d'):
|
||
_existing_tag = '' # 仅 LLM 路径可撤销推荐
|
||
else:
|
||
_existing_tag = _old_tag # 技术路径一律不动 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
|
||
|
||
# ── action 权限保护已在上方信号权威块中统一处理 ──
|
||
|
||
# ── UPSERT（2026-07-23 老爸：新增计算列不该用DELETE+INSERT，该用UPDATE）──
|
||
# 只写本函数拥有的列；rr_low/rr_high(recompute_rr拥有)、superseded_at(data_governance
|
||
# 拥有)、created_at(创建时间)不在写集内 → 天然保留，未来新增计算列自动免疫。
|
||
# 同时消除 DELETE→INSERT 之间崩溃=行丢失的原子性窗口，以及 created_at 被重置的副作用。
|
||
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'),
|
||
?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||
ON CONFLICT(code) DO UPDATE SET
|
||
name=excluded.name, version=excluded.version, price=excluded.price,
|
||
cost=excluded.cost, shares=excluded.shares,
|
||
stop_loss=excluded.stop_loss, take_profit=excluded.take_profit,
|
||
entry_low=excluded.entry_low, entry_high=excluded.entry_high,
|
||
currency=excluded.currency, strategy_type=excluded.strategy_type,
|
||
action=excluded.action, timing_signal=excluded.timing_signal,
|
||
rr_ratio=excluded.rr_ratio, tech_snapshot=excluded.tech_snapshot,
|
||
stock_category=excluded.stock_category, sector_context=excluded.sector_context,
|
||
status=excluded.status, trigger_json=excluded.trigger_json,
|
||
changelog_json=excluded.changelog_json, source=excluded.source,
|
||
reason=excluded.reason, updated_at=excluded.updated_at,
|
||
avg_price=excluded.avg_price, decision_timestamp=excluded.decision_timestamp,
|
||
note=excluded.note, quality_check=excluded.quality_check,
|
||
quality_checked_at=excluded.quality_checked_at,
|
||
quality_issues_json=excluded.quality_issues_json,
|
||
position_advice=excluded.position_advice,
|
||
signal_factors_json=excluded.signal_factors_json,
|
||
time_horizon=excluded.time_horizon, decision_type=excluded.decision_type,
|
||
full_analysis=excluded.full_analysis, reassessed_at=excluded.reassessed_at,
|
||
tag=excluded.tag
|
||
""", (
|
||
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()
|
||
# ── 策略版本追踪（每次策略写入后自动记录，2026-07-27）──
|
||
if _existing_tag == 'current_recommend':
|
||
track_strategy_version(conn, code)
|
||
# ── 推荐转场：LLM路径新转为 current_recommend → 记入摘要队列（不逐只推送）──
|
||
if _existing_tag == 'current_recommend' and _old_tag != 'current_recommend' \
|
||
and source_trigger in ('batch_12d', 'per_stock_12d'):
|
||
enqueue_recommend(conn, code)
|
||
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()
|
||
# ── 同步 holding_strategies（2026-07-27 老爸：导入持仓后盯盘应即时出现）──
|
||
for h in holdings:
|
||
code = h.get('code')
|
||
name = h.get('name', '')
|
||
shares = h.get('shares') or 0
|
||
if not code or shares <= 0:
|
||
continue
|
||
existing = conn.execute(
|
||
"SELECT id, decision_type FROM holding_strategies WHERE code=? AND status='active'",
|
||
(code,)).fetchone()
|
||
if not existing:
|
||
# 全新持仓：创建基础策略条目
|
||
conn.execute("""
|
||
INSERT INTO holding_strategies (code, name, decision_type, strategy_type,
|
||
status, timing_signal, created_at)
|
||
VALUES (?, ?, '持仓策略', 'holding', 'active', '关注', datetime('now','localtime'))
|
||
""", (code, name))
|
||
elif existing[1] != '持仓策略':
|
||
# 已有条目但类型不对（自选转持仓）
|
||
conn.execute(
|
||
"UPDATE holding_strategies SET decision_type='持仓策略' WHERE code=? AND status='active'",
|
||
(code,))
|
||
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）
|
||
|
||
2026-08-11 修复: 原 DELETE 全表 + INSERT 在大库(mofin.db 6GB)上
|
||
持写锁过久，导致并发 cron (candidate_filter 等) database is locked。
|
||
改为固定 id=1 单行 upsert，锁持有时间从秒级降至毫秒级。
|
||
"""
|
||
import json
|
||
conn.execute(
|
||
"INSERT INTO capital_flow_cache (id, cache_json, updated_at) VALUES (1,?,datetime('now','localtime')) "
|
||
"ON CONFLICT(id) DO UPDATE SET cache_json=excluded.cache_json, updated_at=excluded.updated_at",
|
||
(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 WHERE id=1 ORDER BY updated_at DESC LIMIT 1").fetchone()
|
||
return json.loads(r['cache_json']) if r else {}
|