feat: 策略研究Tab全流程回测框架 + API端点 + 前端渲染 + 盘中大跌分析日志

This commit is contained in:
hmo
2026-07-28 22:53:29 +08:00
parent d3298dbf83
commit 63b53f5c56
8 changed files with 7669 additions and 1738 deletions
+21
View File
@@ -310,3 +310,24 @@ cron prompt for 系统健康检查-开盘前 (job_id: 37c02f4d7df9)
- **总资产**: 896,562
- **仓位**: 73.08%
- **截图头部659,354与xls的655,231差异来源**: 截图是13:01价格,xls是23:30价格,中间价格波动导致
## 2026-07-28 13:51 盘中系统性大跌应对 — 韩国二次熔断+亚太科技股暴跌
**发现了什么:**
今天KOSPI两度触发熔断(一度跌超8%),日经-4%,MSCI亚太较6月高点跌10%,科创50-5.52%/创业板-6.55%。但上证仅-1.35%、恒指-0.34%、比亚迪/腾讯逆势上涨。
**关键判断:**
- 定性为板块冲击(B类),非全面系统性。上证/恒指有韧性是关键区分信号
- 比亚迪+0.51%、腾讯+0.23%逆势上涨验证了A/H股科技脱钩叙事
- 信号id=1441已处理(processed=1),自愈执行器报警的是状态文件过期导致的重复触发
**止损距扫描结果:**
- 中芯国际(688981)现价138.41,跌破旧止损141.21(-2.0%),但策略状态为closed,需发布新策略
- 丘钛科技(01478)距止损仅1.2%,临界
- 中国神华(01088)距止损仅1.6%,临界
- 中际旭创(300308)距止损5.2%,策略已closed
**萃取知识:**
1. KOSPI熔断+日韩暴跌传导至A股科技股,但上证/恒指未破位=非系统性+板块冲击。判断A股的不可一概而论。
2. 比亚迪、腾讯在科技暴跌日逆势上涨,说明个股基本面>板块情绪。不能简单用"科技股暴跌"标签覆盖所有科技持仓。
3. 自愈执行器重复报警的根本原因是state文件expired后未及时更新,导致管道认为这是未处理的新问题。修复措施:更新state文件。
+488
View File
@@ -0,0 +1,488 @@
#!/usr/bin/env python3
"""MoFin 回测框架 v2 — 从 stock_daily 原始行情计算全部指标,模拟全流程"""
import sqlite3, json, math
from datetime import datetime, timedelta
from collections import defaultdict
DB_PATH = "/home/hmo/MoFin/data/mofin.db"
# ── 技术指标计算(纯 Python,不依赖第三方库)──
def calc_ma(series, n):
"""简单移动平均"""
result = []
for i in range(len(series)):
if i < n - 1:
result.append(None)
else:
result.append(sum(series[i-n+1:i+1]) / n)
return result
def calc_ema(series, n):
"""指数移动平均"""
result = []
k = 2 / (n + 1)
for i in range(len(series)):
if i == 0:
result.append(series[i])
else:
result.append(series[i] * k + result[-1] * (1 - k))
return result
def calc_rsi(series, n=14):
"""RSI"""
deltas = [series[i] - series[i-1] for i in range(1, len(series))]
gains = [d if d > 0 else 0 for d in deltas]
losses = [-d if d < 0 else 0 for d in deltas]
result = [None] * (n + 1)
avg_gain = sum(gains[:n]) / n
avg_loss = sum(losses[:n]) / n
if avg_loss == 0:
result.append(100)
else:
rs = avg_gain / avg_loss
result.append(100 - 100 / (1 + rs))
for i in range(n, len(gains)):
avg_gain = (avg_gain * (n-1) + gains[i]) / n
avg_loss = (avg_loss * (n-1) + losses[i]) / n
if avg_loss == 0:
result.append(100)
else:
rs = avg_gain / avg_loss
result.append(100 - 100 / (1 + rs))
# Pad front to match original length
while len(result) < len(series):
result.insert(0, None)
return result[:len(series)]
def calc_macd(series, fast=12, slow=26, signal=9):
"""MACD: 返回 macd_line, signal_line, histogram"""
ema_fast = calc_ema(series, fast)
ema_slow = calc_ema(series, slow)
macd_line = [ema_fast[i] - ema_slow[i] if ema_fast[i] is not None and ema_slow[i] is not None else None for i in range(len(series))]
# Recompute signal line from macd_line (only non-None values)
macd_clean = [v for v in macd_line if v is not None]
if not macd_clean:
return [None]*len(series), [None]*len(series), [None]*len(series)
signal_raw = calc_ema(macd_clean, signal)
signal_line = [None] * (len(series) - len(macd_clean)) + signal_raw
histogram = [macd_line[i] - signal_line[i] if macd_line[i] is not None and signal_line[i] is not None else None for i in range(len(series))]
return macd_line, signal_line, histogram
def calc_tr(highs, lows, closes):
"""True Range"""
tr = [highs[0] - lows[0]]
for i in range(1, len(highs)):
hl = highs[i] - lows[i]
hc = abs(highs[i] - closes[i-1])
lc = abs(lows[i] - closes[i-1])
tr.append(max(hl, hc, lc))
return tr
def calc_atr(highs, lows, closes, n=14):
"""ATR"""
tr = calc_tr(highs, lows, closes)
return calc_ema(tr, n)
def calc_trend_strength(highs, lows, closes, n=14):
"""简易趋势强度(替代 ADX: (DI+ - DI-) / (DI+ + DI-) 归一化"""
tr = calc_tr(highs, lows, closes)
up = [highs[i] - highs[i-1] for i in range(1, len(highs))]
down = [lows[i-1] - lows[i] for i in range(1, len(lows))]
# Directional indicators (raw)
di_plus_raw = [0.0] * len(up)
di_minus_raw = [0.0] * len(up)
for i in range(len(up)):
if up[i] > down[i] and up[i] > 0:
di_plus_raw[i] = up[i]
if down[i] > up[i] and down[i] > 0:
di_minus_raw[i] = down[i]
# Normalize by TR
tr_val = tr[i+1] if (i+1) < len(tr) else tr[-1]
if tr_val > 0:
di_plus_raw[i] = di_plus_raw[i] / tr_val * 100
di_minus_raw[i] = di_minus_raw[i] / tr_val * 100
# SMA of DMI
result = []
for i in range(len(di_plus_raw)):
if i < n - 1:
result.append(None)
else:
avg_plus = sum(di_plus_raw[i-n+1:i+1]) / n
avg_minus = sum(di_minus_raw[i-n+1:i+1]) / n
if avg_plus + avg_minus > 0:
dx = abs(avg_plus - avg_minus) / (avg_plus + avg_minus) * 100
else:
dx = 0
result.append(dx)
# Pad front to match original highs length
return [None] * (len(highs) - len(result)) + result
def calc_obv(closes, volumes):
"""OBV"""
obv = [volumes[0]]
for i in range(1, len(closes)):
if closes[i] > closes[i-1]:
obv.append(obv[-1] + volumes[i])
elif closes[i] < closes[i-1]:
obv.append(obv[-1] - volumes[i])
else:
obv.append(obv[-1])
return obv
def calc_roc(series, n=10):
"""Rate of Change"""
result = []
for i in range(len(series)):
if i < n:
result.append(None)
else:
result.append((series[i] - series[i-n]) / series[i-n] * 100 if series[i-n] != 0 else 0)
return result
def prepare_bars(code, start_date, end_date):
"""获取原始数据并计算全部指标"""
conn = sqlite3.connect(DB_PATH)
rows = conn.execute("""
SELECT date, open, close, high, low, volume, amount
FROM stock_daily WHERE code=? AND date>=? AND date<=?
ORDER BY date
""", (code, start_date, end_date)).fetchall()
conn.close()
if not rows or len(rows) < 30:
return None
dates = [r[0] for r in rows]
opens = [r[1] for r in rows]
closes = [r[2] for r in rows]
highs = [r[3] for r in rows]
lows = [r[4] for r in rows]
volumes = [r[5] for r in rows]
# 计算全部指标
ma5 = calc_ma(closes, 5)
ma10 = calc_ma(closes, 10)
ma20 = calc_ma(closes, 20)
ma60 = calc_ma(closes, 60)
rsi = calc_rsi(closes)
macd_line, signal_line, macd_hist = calc_macd(closes)
atr = calc_atr(highs, lows, closes)
trend_strength = calc_trend_strength(highs, lows, closes)
obv = calc_obv(closes, volumes)
roc = calc_roc(closes)
bars = []
for i in range(len(dates)):
bars.append({
'date': dates[i],
'open': opens[i],
'close': closes[i],
'high': highs[i],
'low': lows[i],
'volume': volumes[i],
'ma5': ma5[i] if i < len(ma5) else None,
'ma10': ma10[i] if i < len(ma10) else None,
'ma20': ma20[i] if i < len(ma20) else None,
'ma60': ma60[i] if i < len(ma60) else None,
'rsi': rsi[i] if i < len(rsi) else None,
'macd': macd_line[i] if i < len(macd_line) else None,
'macd_signal': signal_line[i] if i < len(signal_line) else None,
'macd_hist': macd_hist[i] if i < len(macd_hist) else None,
'atr': atr[i] if i < len(atr) else None,
'adx': trend_strength[i] if i < len(trend_strength) else None,
'obv': obv[i] if i < len(obv) else None,
'roc': roc[i] if i < len(roc) else None,
})
return bars
def compute_single_score(bars):
"""对单个股票计算多因子评分(与 MoFin 现有一致)"""
if not bars or len(bars) < 15:
return None
last = bars[-1]
prev5 = bars[-5] if len(bars) >= 5 else bars[-2] if len(bars) >= 2 else last
prev10 = bars[-10] if len(bars) >= 10 else bars[0]
def v(val): return val if val is not None else 0
score = {'trend': 0, 'momentum': 0, 'volume': 0, 'volatility': 0, 'risk': 0}
# 趋势分 (0-25)
ts = 0
adx = v(last.get('adx'))
if adx > 25: ts += 10
elif adx > 20: ts += 5
mh = v(last.get('macd_hist'))
pmh = v(prev5.get('macd_hist'))
if mh > 0 and mh > pmh: ts += 10
elif mh > 0: ts += 5
elif mh < 0 and mh < pmh: ts -= 5
ma5 = v(last.get('ma5'))
ma10 = v(last.get('ma10'))
ma20 = v(last.get('ma20'))
if ma5 > ma10 > ma20 and ma5 > 0 and ma10 > 0: ts += 5
elif ma5 < ma10 < ma20 and ma5 > 0 and ma10 > 0: ts -= 5
score['trend'] = max(0, min(25, ts))
# 动量分 (0-20)
ms = 0
roc = v(last.get('roc'))
if roc > 5: ms += 5
elif roc > 2: ms += 2
elif roc < -5: ms -= 5
elif roc < -2: ms -= 2
rsi = v(last.get('rsi'))
if 30 <= rsi <= 70: ms += 5
if 50 < rsi <= 65: ms += 3
elif rsi > 70: ms -= 3
elif rsi < 30: ms -= 3
close = v(last.get('close'))
if ma20 > 0:
pct = (close - ma20) / ma20 * 100
if -3 <= pct <= 5: ms += 7
elif pct < -10: ms -= 3
score['momentum'] = max(0, min(20, ms))
# 量能分 (0-20)
vs = 0
obv_cur = v(last.get('obv'))
obv_prev = v(prev5.get('obv'))
vol_cur = v(last.get('volume'))
vol_prev = v(prev5.get('volume'))
if obv_cur > obv_prev and vol_cur > 0 and vol_prev > 0:
vr = vol_cur / vol_prev
if vr > 1.5: vs += 10
elif vr > 1.2: vs += 5
if obv_cur > obv_prev: vs += 5
if vol_prev > 0 and vol_cur / vol_prev < 0.5: vs -= 5
score['volume'] = max(0, min(20, vs))
# 波动率分 (0-15)
vls = 5
atr = v(last.get('atr'))
if close > 0 and atr > 0:
atr_pct = atr / close * 100
if 1.5 <= atr_pct <= 3.5: vls += 5
elif atr_pct < 1: vls += 3
elif atr_pct > 5: vls -= 5
if len(bars) >= 20:
cls = [v(b.get('close')) for b in bars[-20:]]
if min(cls) > 0:
vol_range = (max(cls) - min(cls)) / min(cls) * 100
if vol_range < 10: vls += 5
elif vol_range > 30: vls -= 5
score['volatility'] = max(0, min(15, vls))
# 风险分 (0-20)
rs = 10
if len(bars) >= 10:
low5 = min(bars[-5:], key=lambda x: v(x.get('low')))
low10 = min(bars[-10:-5], key=lambda x: v(x.get('low'))) if len(bars) >= 10 else low5
if v(low5.get('close')) < v(low10.get('close')) and v(low5.get('macd_hist')) > v(low10.get('macd_hist')):
rs += 5
if v(low5.get('close')) > v(low10.get('close')) and v(low5.get('macd_hist')) < v(low10.get('macd_hist')):
rs -= 5
if ma20 > 0:
dist = (close - ma20) / ma20 * 100
if dist < 2: rs += 5
elif dist > 15: rs -= 5
score['risk'] = max(0, min(20, rs))
total = sum(score.values())
return total, score
def compute_kelly(total_score, target_pct, stop_pct):
"""半 Kelly 仓位"""
p = min(0.6, total_score / 100)
if stop_pct <= 0:
return 0
b = target_pct / stop_pct
if b <= 1:
return 0
kelly = (p * b - (1 - p)) / b
return max(0, min(0.5, kelly)) * 0.5 # 半 Kelly
def run_strategy_research(start_date, end_date, capital=1000000):
"""主回测入口"""
conn = sqlite3.connect(DB_PATH)
# 1. 获取所有有历史数据的股票
stocks = conn.execute("""
SELECT DISTINCT sd.code, COALESCE(s.name, sd.code) as name
FROM stock_daily sd
LEFT JOIN stocks s ON sd.code = s.code
WHERE sd.date>=? AND sd.date<=?
""", (start_date, end_date)).fetchall()
results = {
'period': f"{start_date} ~ {end_date}",
'capital': capital,
'total_stocks_screened': len(stocks),
'scored_stocks': 0,
'buy_signals': 0,
'trades': [],
'summary': {}
}
trades = []
code_names = {}
eval_dates = [] # collect evaluation dates
for code, name in stocks:
code_names[code] = name
bars = prepare_bars(code, start_date, end_date)
if not bars or len(bars) < 20:
continue
# 2. 滚动评估:每周评分一次
window_start = 20 # need at least 20 bars for indicators
while window_start < len(bars):
window = bars[:window_start]
score_result = compute_single_score(window)
if score_result is None:
window_start += 5
continue
total_score, score_components = score_result
results['scored_stocks'] += 1
last = window[-1]
close = v_close = last.get('close', 0) or 0
ma20 = last.get('ma20', 0) or 0
# 3. 买入条件:总评分 >= 45 且 momentum >= 8
buy_signal = total_score >= 45 and score_components['momentum'] >= 8
if buy_signal:
results['buy_signals'] += 1
entry_price = close
atr_val = last.get('atr', 0) or 0
if atr_val > 0:
target_price = entry_price * 1.10
stop_price = entry_price - atr_val * 2.0
else:
target_price = entry_price * 1.10
stop_price = entry_price * 0.93
kelly = compute_kelly(total_score, 0.10, (entry_price - stop_price) / entry_price)
# 4. 模拟未来走势
future = bars[window_start:min(window_start+20, len(bars))]
exit_price = None
exit_reason = None
max_high = entry_price
min_low = entry_price
for fb in future:
fb_high = fb.get('high', 0) or 0
fb_low = fb.get('low', 0) or 0
fb_close = fb.get('close', 0) or 0
max_high = max(max_high, fb_high)
min_low = min(min_low, fb_low)
if fb_high >= target_price:
exit_price = target_price
exit_reason = 'target'
break
elif fb_low <= stop_price:
exit_price = fb_close
exit_reason = 'stop'
break
if exit_price is None:
if future:
exit_price = future[-1].get('close', entry_price)
else:
exit_price = entry_price
exit_reason = 'keep'
hold_days = 0
if exit_reason and exit_reason != 'keep' and future:
for idx, fb in enumerate(future):
fb_high = fb.get('high', 0) or 0
fb_low = fb.get('low', 0) or 0
if (exit_reason == 'target' and fb_high >= target_price) or \
(exit_reason == 'stop' and fb_low <= stop_price):
hold_days = idx + 1
break
elif exit_reason == 'keep':
hold_days = len(future)
profit_pct = (exit_price - entry_price) / entry_price * 100 if entry_price > 0 else 0
trades.append({
'code': code,
'name': name,
'entry_date': last.get('date'),
'entry_price': round(entry_price, 2),
'exit_price': round(exit_price, 2),
'profit_pct': round(profit_pct, 2),
'exit_reason': exit_reason,
'hold_days': hold_days,
'score': total_score,
'kelly': round(kelly, 3),
'stop_loss': round(stop_price, 2),
'target': round(target_price, 2),
'max_high': round(max_high, 2),
'min_low': round(min_low, 2)
})
window_start += 5 # 每 5 个交易日评估一次
conn.close()
# 5. 汇总
if trades:
profits = [t['profit_pct'] for t in trades]
wins = [t for t in trades if t['profit_pct'] > 0]
losses = [t for t in trades if t['profit_pct'] <= 0]
win_rate = len(wins) / len(trades) * 100 if trades else 0
avg_profit = sum(profits) / len(profits) if profits else 0
avg_win = sum(t['profit_pct'] for t in wins) / len(wins) if wins else 0
avg_loss = sum(t['profit_pct'] for t in losses) / len(losses) if losses else 0
mean_ret = avg_profit / 100
if len(profits) > 1:
variance = sum((p/100 - mean_ret)**2 for p in profits) / (len(profits) - 1)
std_ret = math.sqrt(variance)
else:
std_ret = 0
sharpe = mean_ret / std_ret * math.sqrt(252) if std_ret > 0 else 0
# 最大回撤:用实际资本曲线计算
capital_curve = [capital]
for t in trades:
new_cap = capital_curve[-1] * (1 + t['profit_pct'] / 100)
capital_curve.append(new_cap)
peak_cap = capital
max_dd = 0
for c in capital_curve:
peak_cap = max(peak_cap, c)
dd = (peak_cap - c) / peak_cap * 100
max_dd = max(max_dd, dd)
results['trades'] = sorted(trades, key=lambda x: abs(x['profit_pct']), reverse=True)[:50]
results['summary'] = {
'total_trades': len(trades),
'win_rate': round(win_rate, 1),
'avg_profit_pct': round(avg_profit, 2),
'avg_win_pct': round(avg_win, 2),
'avg_loss_pct': round(avg_loss, 2),
'sharpe_ratio': round(sharpe, 2),
'max_drawdown_pct': round(max_dd, 2),
'profit_factor': round(abs(avg_win / avg_loss), 2) if avg_loss != 0 else float('inf'),
'wins': len(wins),
'losses': len(losses),
'capital_end': round(capital * (1 + avg_profit/100), 2)
}
return results
if __name__ == '__main__':
end = datetime.now().strftime('%Y-%m-%d')
start = (datetime.now() - timedelta(days=180)).strftime('%Y-%m-%d')
r = run_strategy_research(start, end)
print(json.dumps(r, indent=2, ensure_ascii=False))
+316
View File
@@ -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")
File diff suppressed because it is too large Load Diff
+316
View File
@@ -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")
+2341
View File
File diff suppressed because it is too large Load Diff
+1761 -1738
View File
File diff suppressed because it is too large Load Diff
+85
View File
@@ -70,6 +70,7 @@ body { background: #0f0f13; color: #e2e8f0; }
<button class="tab-btn px-4 py-2 text-sm rounded-lg" data-tab="signals">🔍 信号</button>
<button class="tab-btn px-4 py-2 text-sm rounded-lg" data-tab="health">🏥 健康</button>
<button class="tab-btn px-4 py-2 text-sm rounded-lg" data-tab="principles">📐 开发原则</button>
<button class="tab-btn px-4 py-2 text-sm rounded-lg" data-tab="research">📋 研究</button>
<a href="/upload" class="tab-btn px-4 py-2 text-sm rounded-lg" style="text-decoration:none;color:#fbbf24;border-color:#fbbf24">📸 上传</a>
</div>
@@ -82,6 +83,7 @@ body { background: #0f0f13; color: #e2e8f0; }
<div id="tab-market" class="tab-content hidden"></div>
<div id="tab-signals" class="tab-content hidden"></div>
<div id="tab-health" class="tab-content hidden"></div>
<div id="tab-research" class="tab-content hidden"></div>
<div id="tab-principles" class="tab-content hidden">
<div class="flex gap-1 mb-4 bg-slate-900/50 rounded-xl p-1 border border-slate-800/50" id="princBar">
<button class="princ-btn active px-4 py-1.5 text-sm rounded-lg" data-princ="spec">G 规范</button>
@@ -258,6 +260,7 @@ function renderTab(name) {
else if (name === 'health') renderHealth();
else if (name === 'principles') renderPrinciples();
else if (name === 'watch') renderWatch();
else if (name === 'research') renderResearch();
}
// ── Data Fetching ──
@@ -1917,6 +1920,88 @@ function inMd(t) {
.replace(/`([^`]+)`/g, '<code class="bg-slate-900 px-1 rounded text-xs font-mono">$1</code>')
.replace(/\[([^\]]+)\]\(([^)]+)\)/g, '<a href="$2" class="text-[#58a6ff]" target="_blank">$1</a>');
}
function renderResearch() {
const el = document.getElementById('tab-research');
if (!el) return;
el.innerHTML = '<div class="flex gap-2 items-center mb-4"><h2 class="text-lg font-bold">📋 策略研究 — 全流程回测</h2>' +
'<select id="btPeriod" class="bg-slate-800 text-sm rounded-lg px-2 py-1 border border-slate-700" onchange="runBacktest()">' +
'<option value="1m">近 1 个月</option>' +
'<option value="6m" selected>近 6 个月</option>' +
'<option value="1y">近 1 年</option>' +
'<option value="2y">近 2 年</option>' +
'</select>' +
'<button onclick="runBacktest()" class="px-3 py-1 bg-blue-600 hover:bg-blue-500 rounded-lg text-sm">▶ 运行</button></div>' +
'<div id="btResults" class="text-slate-400 text-sm">点击「运行」开始回测验证</div>';
window.btEl = el;
}
async function runBacktest() {
const period = document.getElementById('btPeriod')?.value || '6m';
const resultsEl = document.getElementById('btResults');
if (!resultsEl) return;
resultsEl.innerHTML = '<div class="text-blue-400 animate-pulse">⏳ 回测运行中(约 15-30 秒)...</div>';
try {
const resp = await fetch('/api/research/backtest?period=' + period);
const data = await resp.json();
if (data.error) { resultsEl.innerHTML = '<div class="text-red-400">错误: ' + data.error + '</div>'; return; }
displayBacktestResults(data);
} catch(e) {
resultsEl.innerHTML = '<div class="text-red-400">请求失败: ' + e.message + '</div>';
}
}
function displayBacktestResults(data) {
const el = document.getElementById('btResults');
if (!el) return;
const s = data.summary || {};
const winRate = s.win_rate || 0;
const winRateClass = winRate >= 50 ? 'text-green-400' : winRate >= 40 ? 'text-yellow-400' : 'text-red-400';
const sharpeClass = (s.sharpe_ratio || 0) >= 1 ? 'text-green-400' : (s.sharpe_ratio || 0) >= 0 ? 'text-yellow-400' : 'text-red-400';
let html = '<div class="grid grid-cols-2 md:grid-cols-4 gap-3 mb-4">' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">测试周期</div><div class="text-sm font-mono">' + (data.period || '') + '</div></div>' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">初始资金</div><div class="text-sm font-mono">¥' + (s.capital_end ? (data.capital || 1000000).toLocaleString() : '100万') + '</div></div>' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">最终资金</div><div class="text-sm font-mono">¥' + (s.capital_end ? s.capital_end.toLocaleString() : '-') + '</div></div>' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">总交易</div><div class="text-sm font-mono">' + (s.total_trades || 0) + ' 笔</div></div>' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">胜率</div><div class="text-lg font-bold ' + winRateClass + '">' + s.win_rate + '%</div></div>' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">平均收益</div><div class="text-lg font-bold ' + ((s.avg_profit_pct || 0) >= 0 ? 'text-green-400' : 'text-red-400') + '">' + (s.avg_profit_pct || 0) + '%</div></div>' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">夏普比率</div><div class="text-lg font-bold ' + sharpeClass + '">' + (s.sharpe_ratio || 0).toFixed(2) + '</div></div>' +
'<div class="bg-slate-800/50 rounded-lg p-3"><div class="text-xs text-slate-500">最大回撤</div><div class="text-lg font-bold text-red-400">' + (s.max_drawdown_pct || 0).toFixed(1) + '%</div></div>' +
'</div>' +
'<div class="grid grid-cols-4 gap-2 mb-4 text-xs">' +
'<div class="bg-green-900/20 rounded p-2"><span class="text-green-400">✅ 盈利: ' + (s.wins || 0) + '</span> <span class="text-slate-500">均+' + (s.avg_win_pct || 0).toFixed(1) + '%</span></div>' +
'<div class="bg-red-900/20 rounded p-2"><span class="text-red-400">❌ 亏损: ' + (s.losses || 0) + '</span> <span class="text-slate-500">均' + (s.avg_loss_pct || 0).toFixed(1) + '%</span></div>' +
'<div class="bg-slate-800/30 rounded p-2"><span class="text-yellow-400">📊 盈亏比: ' + (s.profit_factor || 0).toFixed(2) + '</span></div>' +
'<div class="bg-slate-800/30 rounded p-2"><span class="text-slate-300">📈 筛选: ' + (data.total_stocks_screened || 0) + ' 只</span></div>' +
'</div>';
// Trade list
const trades = data.trades || [];
html += '<div class="mb-2 text-sm font-bold text-slate-300">交易明细(前 50 笔)</div>' +
'<div class="overflow-x-auto"><table class="w-full text-xs"><thead><tr class="text-slate-500 border-b border-slate-700">' +
'<th class="text-left px-2 py-1">代码</th><th class="text-left px-2 py-1">名称</th><th class="text-left px-2 py-1">买入日</th>' +
'<th class="text-right px-2 py-1">买入价</th><th class="text-right px-2 py-1">卖出价</th><th class="text-right px-2 py-1">收益</th>' +
'<th class="text-center px-2 py-1">结果</th><th class="text-right px-2 py-1">持仓</th><th class="text-right px-2 py-1">评分</th></tr></thead><tbody>';
for (const t of trades) {
const pnlClass = (t.profit_pct || 0) >= 0 ? 'text-green-400' : 'text-red-400';
const reasonMap = { 'target': '🎯 止盈', 'stop': '🛑 止损', 'keep': '⏳ 持仓中' };
html += '<tr class="border-b border-slate-800/50 hover:bg-slate-800/30">' +
'<td class="px-2 py-1 font-mono">' + (t.code || '') + '</td>' +
'<td class="px-2 py-1">' + (t.name || '') + '</td>' +
'<td class="px-2 py-1 font-mono">' + (t.entry_date || '') + '</td>' +
'<td class="text-right px-2 py-1 font-mono">' + (t.entry_price || 0).toFixed(2) + '</td>' +
'<td class="text-right px-2 py-1 font-mono">' + (t.exit_price || 0).toFixed(2) + '</td>' +
'<td class="text-right px-2 py-1 font-mono ' + pnlClass + '">' + (t.profit_pct || 0).toFixed(1) + '%</td>' +
'<td class="text-center px-2 py-1">' + (reasonMap[t.exit_reason] || t.exit_reason || '') + '</td>' +
'<td class="text-right px-2 py-1">' + (t.hold_days || 0) + 'd</td>' +
'<td class="text-right px-2 py-1">' + (t.score || 0) + '</td></tr>';
}
html += '</tbody></table></div>';
el.innerHTML = html;
}
</script>
</body>
</html>