Hugging Face's logo

Spaces:
sgfw
/
kronos
Running

App Files Community
kronos / webui /app.py
sgfw's picture
sgfw
Upload 82 files
1d900df verified
Raw History Blame Contribute Delete
168 kB
"""
Kronos Enhanced Web UI - Backend
集成波浪理论、技术指标、量价分析的增强版预测系统
"""
import os
import sys
import json
import pandas as pd
import numpy as np
import plotly.graph_objects as go
import plotly.utils
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')
from flask import Flask, make_response, render_template, request, jsonify
from flask_cors import CORS
# import akshare as ak # 已弃用，改用腾讯财经 API
from io import StringIO
import threading
import time as time_module
# Classic pattern detection
try:
import sys, importlib, types as _types
_cp_path = os.path.join(os.path.dirname(__file__), '..', 'model', 'classic_patterns.py')
if os.path.exists(_cp_path) and 'classic_patterns' not in sys.modules:
_spec = importlib.util.spec_from_file_location('classic_patterns', os.path.abspath(_cp_path))
_mod = importlib.util.module_from_spec(_spec)
sys.modules['classic_patterns'] = _mod
_spec.loader.exec_module(_mod)
from classic_patterns import detect_classic_patterns
CLASSIC_PATTERN_AVAILABLE = True
print("[OK] classic_patterns module loaded")
else:
CLASSIC_PATTERN_AVAILABLE = False
except Exception as _e:
CLASSIC_PATTERN_AVAILABLE = False
print(f"[Warning] classic_patterns not available: {_e}")
# Load Wyckoff VPA module (独立于 enhanced modules)
WYCKOFF_VPA_AVAILABLE = False
try:
_wv_path = os.path.join(os.path.dirname(__file__), '..', 'model', 'wyckoff_vpa.py')
if os.path.exists(_wv_path) and 'wyckoff_vpa' not in sys.modules:
_wv_spec = importlib.util.spec_from_file_location('wyckoff_vpa', os.path.abspath(_wv_path))
_wv_mod = importlib.util.module_from_spec(_wv_spec)
sys.modules['wyckoff_vpa'] = _wv_mod
_wv_spec.loader.exec_module(_wv_mod)
if 'wyckoff_vpa' in sys.modules:
from wyckoff_vpa import detect_wyckoff_vpa
WYCKOFF_VPA_AVAILABLE = True
except Exception as _wve:
WYCKOFF_VPA_AVAILABLE = False
print(f"[Warning] wyckoff_vpa not available: {_wve}")
# Wave Analysis Module (dynamic load, same pattern as classic_patterns/wyckoff_vpa)
WAVE_ANALYSIS_AVAILABLE = False
try:
_wa_path = os.path.join(os.path.dirname(__file__), '..', 'model', 'wave_analysis.py')
if os.path.exists(_wa_path) and 'wave_analysis' not in sys.modules:
_wa_spec = importlib.util.spec_from_file_location('wave_analysis', os.path.abspath(_wa_path))
_wa_mod = importlib.util.module_from_spec(_wa_spec)
sys.modules['wave_analysis'] = _wa_mod
_wa_spec.loader.exec_module(_wa_mod)
if 'wave_analysis' in sys.modules:
from wave_analysis import analyze_wave_patterns, WaveAnalyzer
WAVE_ANALYSIS_AVAILABLE = True
print('[OK] wave_analysis module loaded')
else:
print('[Warning] wave_analysis not in sys.modules after load attempt')
except Exception as _we:
WAVE_ANALYSIS_AVAILABLE = False
print(f'[Warning] wave_analysis module not available: {_we}')
# Fundamental Analysis Module (dynamic load)
FUNDAMENTAL_AVAILABLE = False
try:
_fa_path = os.path.join(os.path.dirname(__file__), '..', 'model', 'fundamental_analysis.py')
if os.path.exists(_fa_path) and 'fundamental_analysis' not in sys.modules:
_fa_spec = importlib.util.spec_from_file_location('fundamental_analysis', os.path.abspath(_fa_path))
_fa_mod = importlib.util.module_from_spec(_fa_spec)
sys.modules['fundamental_analysis'] = _fa_mod
_fa_spec.loader.exec_module(_fa_mod)
if 'fundamental_analysis' in sys.modules:
from fundamental_analysis import run_fundamental_analysis
FUNDAMENTAL_AVAILABLE = True
print('[OK] fundamental_analysis module loaded')
else:
print('[Warning] fundamental_analysis not in sys.modules after load attempt')
except Exception as _fe:
FUNDAMENTAL_AVAILABLE = False
print(f'[Warning] fundamental_analysis module not available: {_fe}')
# ====== 踏浪之道 · 主升浪识别模块 (v4.1 评分制) ======
TALANG_AVAILABLE = False
try:
_tl_path = os.path.join(os.path.dirname(__file__), '..', 'model', 'talang_integration.py')
if os.path.exists(_tl_path) and 'talang_integration' not in sys.modules:
_tl_spec = importlib.util.spec_from_file_location('talang_integration', os.path.abspath(_tl_path))
_tl_mod = importlib.util.module_from_spec(_tl_spec)
sys.modules['talang_integration'] = _tl_mod
_tl_spec.loader.exec_module(_tl_mod)
if 'talang_integration' in sys.modules:
from talang_integration import talang_score_analysis, talang_backtest_full, RULE_NAMES
TALANG_AVAILABLE = True
print('[OK] 踏浪之道(talang_integration) module loaded')
else:
print('[Warning] talang_integration not in sys.modules')
except Exception as _tle:
TALANG_AVAILABLE = False
print(f'[Warning] 踏浪之道 not available: {_tle}')
# Add project root directory to path
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, PROJECT_ROOT)
# ---- Mock talib with pure-pandas compat layer ----
import sys, importlib, types as _types
_talib_compat = os.path.join(os.path.dirname(__file__), '..', 'model', 'talib.py')
if os.path.exists(_talib_compat) and 'talib' not in sys.modules:
_spec = importlib.util.spec_from_file_location('talib', os.path.abspath(_talib_compat))
_mod = importlib.util.module_from_spec(_spec)
sys.modules['talib'] = _mod
_spec.loader.exec_module(_mod)
print("[OK] talib compat layer loaded")
app = Flask(__name__)
CORS(app)
# Patch Flask JSON provider to handle numpy types
import flask.json.provider
_orig_default = flask.json.provider.DefaultJSONProvider.default
def _patched_default(self, o):
if isinstance(o, (np.bool_,)):
return bool(o)
if isinstance(o, (np.integer,)):
return int(o)
if isinstance(o, (np.floating,)):
return float(o)
if isinstance(o, np.ndarray):
return o.tolist()
return _orig_default(self, o)
flask.json.provider.DefaultJSONProvider.default = _patched_default
# ---- Global state ----
MODEL_AVAILABLE = False
try:
from model import Kronos, KronosTokenizer, KronosPredictor
MODEL_AVAILABLE = True
except ImportError:
print("[Warning] Kronos model cannot be imported")
# Enhanced modules (direct import to avoid model/__init__.py torch dependency)
ENHANCED_AVAILABLE = False
TechnicalIndicators = VolumePriceAnalyzer = WaveTheoryAnalyzer = MultiDimensionSignalAnalyzer = None
EnhancedKronosPredictor = create_enhanced_predictor = EnhancedKronosConfig = None
try:
import importlib.util, sys, os as _os
_ti_path = _os.path.join(_os.path.dirname(__file__), '..', 'model', 'technical_indicators.py')
_spec = importlib.util.spec_from_file_location('model_technical_indicators', _ti_path)
_ti_mod = importlib.util.module_from_spec(_spec)
sys.modules['model_technical_indicators'] = _ti_mod
_spec.loader.exec_module(_ti_mod)
TechnicalIndicators = _ti_mod.TechnicalIndicators
VolumePriceAnalyzer = _ti_mod.VolumePriceAnalyzer
WaveTheoryAnalyzer = _ti_mod.WaveTheoryAnalyzer
MultiDimensionSignalAnalyzer = _ti_mod.MultiDimensionSignalAnalyzer
print("[OK] technical_indicators loaded (direct)")
except Exception as e:
print(f"[Warning] technical_indicators not available: {e}")
try:
from model.enhanced_kronos import EnhancedKronosPredictor, create_enhanced_predictor
from model.enhanced_config import EnhancedKronosConfig
ENHANCED_AVAILABLE = True
print("[OK] enhanced_kronos loaded")
except ImportError as e:
print(f"[Warning] enhanced_kronos not available: {e}")
# Enable ENHANCED_AVAILABLE if at least technical indicators loaded
if TechnicalIndicators is not None:
ENHANCED_AVAILABLE = True
# Analyzers (stateless, create once)
technical_analyzer = TechnicalIndicators() if TechnicalIndicators else None
volume_price_analyzer = VolumePriceAnalyzer() if VolumePriceAnalyzer else None
wave_analyzer = WaveTheoryAnalyzer() if WaveTheoryAnalyzer else None
multidim_analyzer = MultiDimensionSignalAnalyzer() if MultiDimensionSignalAnalyzer else None
# Model instances
tokenizer = None
model = None
predictor = None
enhanced_predictor = None
# ---- Stock data cache (stock_code -> DataFrame) ----
stock_data_cache = {}
def _download_stock_data(stock_code, start_date='20200101', end_date=None, max_retries=3):
"""Download stock data via Tencent Finance API. Supports A-shares (6-digit) and HK stocks (5-digit)."""
if not end_date:
end_date = datetime.now().strftime('%Y-%m-%d')
# Normalize date format
start_fmt = start_date[:4] + '-' + start_date[4:6] + '-' + start_date[6:8] if len(start_date) == 8 else start_date
# Detect market type
# Strip existing sh/sz prefix first
_clean = stock_code
if stock_code[:2].lower() in ('sh', 'sz') and len(stock_code) > 2:
_clean = stock_code[2:]
is_hk = len(_clean) == 5 and _clean.isdigit()
cache_key = f"{stock_code}_{start_fmt}_{end_date}"
if cache_key in stock_data_cache:
return stock_data_cache[cache_key], None
import urllib.request
last_err = None
for attempt in range(max_retries):
try:
if is_hk:
# HK stock: hkfqkline endpoint, 5-digit code, fields: date/open/close/high/low/volume
param_code = f"hk{_clean}"
url = f'https://web.ifzq.gtimg.cn/appstock/app/hkfqkline/get?param={param_code},day,,,800,qfq'
req = urllib.request.Request(url, headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
resp = urllib.request.urlopen(req, timeout=15)
raw = resp.read().decode()
# Strip JSONP wrapper if present
if raw.startswith('kline_dayqfq'):
raw = raw[raw.index('(') + 1:raw.rindex(')')]
data_json = json.loads(raw)
stock_data = data_json.get('data', {}).get(param_code, {})
klines = stock_data.get('qfqday') or stock_data.get('day', [])
code_label = _clean
else:
# A-share: fqkline endpoint, 6-digit code with market prefix
if _clean.startswith('6'):
market = 'sh'
elif _clean.startswith('0') or _clean.startswith('3'):
market = 'sz'
else:
market = 'sh'
param_code = f"{market}{_clean}"
url = f'https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={param_code},day,{start_fmt},{end_date},800,qfq'
req = urllib.request.Request(url, headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
resp = urllib.request.urlopen(req, timeout=15)
data_json = json.loads(resp.read().decode())
stock_data = data_json.get('data', {}).get(param_code, {})
klines = stock_data.get('qfqday') or stock_data.get('day', [])
code_label = f"{market.upper()}{_clean}"
if not klines or not isinstance(klines, list):
return None, f"No data found for stock {stock_code}"
# Parse klines: A-share=[date, open, close, high, low, vol]; HK=[date, open, close, high, low, vol, ...]
records = []
for line in klines:
if isinstance(line, list):
parts = line
else:
parts = str(line).split(',')
if len(parts) >= 6:
records.append({
'timestamps': parts[0],
'open': float(parts[1]),
'close': float(parts[2]),
'high': float(parts[3]),
'low': float(parts[4]),
'volume': float(parts[5])
})
if not records:
return None, f"No data found for stock {stock_code}"
df = pd.DataFrame(records)
df['timestamps'] = pd.to_datetime(df['timestamps'])
df = df.sort_values('timestamps').reset_index(drop=True)
df['amount'] = df['volume'] * df['close']
stock_data_cache[cache_key] = df
return df, None
except Exception as e:
last_err = e
if attempt < max_retries - 1:
time_module.sleep(1 * (attempt + 1))
return None, f"数据下载失败(已重试{max_retries}次): {str(last_err)[:100]}"
def _download_period_data(stock_code, period='m60', count=800):
"""Download K-line data for various periods via Tencent Finance API (replacing EastMoney).
period: 'day'(day), 'week'(week), 'm60'(60min), 'm30'(30min)
Returns DataFrame same format as _download_stock_data.
"""
import urllib.request
if not stock_code.isdigit():
return None, "Invalid stock code"
# Normalize code with market prefix
if stock_code.startswith('6'):
market = 'sh'
elif stock_code.startswith('0') or stock_code.startswith('3'):
market = 'sz'
elif stock_code.startswith('8') or stock_code.startswith('4'):
market = 'sz' # NEEQ
else:
market = 'sh'
param_code = f"{market}{stock_code}"
# Tencent API kline types: day, week, month, 60min, 30min, 15min, 5min, 1min
klt_map = {'m60': '60min', 'm30': '30min', 'week': 'week', 'day': 'day'}
klt = klt_map.get(period, '60min')
url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={param_code},{klt},,,,{count},qfq"
try:
req = urllib.request.Request(url, headers={
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
resp = urllib.request.urlopen(req, timeout=15)
data_json = json.loads(resp.read().decode())
klines = data_json.get('data', {}).get(param_code, {}).get('qfqday', [])
if not klines or not isinstance(klines, list):
return None, f"No data found for {stock_code} (period={klt})"
records = []
for k in klines:
if not isinstance(k, list) or len(k) < 6:
continue
try:
records.append({
'timestamps': k[0],
'open': float(k[1]),
'close': float(k[2]),
'high': float(k[3]),
'low': float(k[4]),
'volume': float(k[5]),
'amount': float(k[6]) if len(k) > 6 else float(k[5]) * float(k[2]),
})
except (ValueError, IndexError):
continue
if not records:
return None, f"No valid records for {stock_code}"
df = pd.DataFrame(records)
df['timestamps'] = pd.to_datetime(df['timestamps'])
df = df.sort_values('timestamps').reset_index(drop=True)
if 'amount' not in df.columns:
df['amount'] = df['volume'] * df['close']
return df, None
except Exception as e:
return None, f"Period data download failed: {str(e)[:100]}"
def _load_csv_file(file_path):
"""Load CSV, ensure OHLCV + timestamps columns, return DataFrame."""
df = pd.read_csv(file_path)
required_cols = ['open', 'high', 'low', 'close']
if not all(c in df.columns for c in required_cols):
return None, f"Missing required columns: {required_cols}"
for col in required_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
if 'volume' in df.columns:
df['volume'] = pd.to_numeric(df['volume'], errors='coerce')
if 'amount' in df.columns:
df['amount'] = pd.to_numeric(df['amount'], errors='coerce')
ts_col = None
for candidate in ('timestamps', 'timestamp', 'date'):
if candidate in df.columns:
ts_col = candidate
break
if ts_col:
df['timestamps'] = pd.to_datetime(ts_col)
else:
df['timestamps'] = pd.date_range(start='2024-01-01', periods=len(df), freq='1H')
df = df.dropna(subset=['open', 'high', 'low', 'close'])
return df, None
def _load_csv_file_for_analysis(file_path):
"""Load CSV and return a clean df with DatetimeIndex for analysis."""
df, err = _load_csv_file(file_path)
if err:
return None, err
df = df.set_index('timestamps').sort_index()
if 'volume' not in df.columns:
df['volume'] = 0
if 'amount' not in df.columns:
df['amount'] = 0
return df, None
def _load_data_files():
"""Scan data directory for CSV files."""
data_dir = os.path.join(PROJECT_ROOT, 'data')
files = []
if os.path.exists(data_dir):
for f in os.listdir(data_dir):
if f.endswith(('.csv', '.feather')):
fp = os.path.join(data_dir, f)
sz = os.path.getsize(fp)
files.append({
'name': f,
'path': fp,
'size': f"{sz/1024:.1f} KB" if sz < 1048576 else f"{sz/1048576:.1f} MB"
})
return files
# ============================================================
# Helper: chart builders
# ============================================================
def _build_kline_chart(df, pred_df, lookback, pred_len,
actual_df=None, start_idx=0, wave_analysis=None):
"""Build a candlestick chart with wave overlays, MA lines on historical data."""
if start_idx + lookback + pred_len <= len(df):
hist_df = df.iloc[start_idx:start_idx + lookback]
else:
avail = min(lookback, len(df) - start_idx)
hist_df = df.iloc[start_idx:start_idx + avail]
fig = go.Figure()
# Historical K-line (convert all to native float)
h_x = [pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in hist_df['timestamps'].tolist()]
h_close = [float(v) for v in hist_df['close'].tolist()]
fig.add_trace(go.Candlestick(
x=h_x,
open=[float(v) for v in hist_df['open'].tolist()],
high=[float(v) for v in hist_df['high'].tolist()],
low=[float(v) for v in hist_df['low'].tolist()],
close=h_close,
name='Historical',
increasing_line_color='#EF5350', decreasing_line_color='#26A69A'
))
# MA moving averages on historical close (SMA 5/10/20/60)
ma_cfg = [
(5, 'MA5', '#FF6B6B', 1.3),
(10, 'MA10', '#4ECDC4', 1.3),
(20, 'MA20', '#FFD93D', 1.3),
(60, 'MA60', '#6C5CE7', 1.3),
]
for period, label, color, width in ma_cfg:
if len(h_close) >= period:
ma_series = pd.Series(h_close).rolling(window=period, min_periods=period).mean()
ma_vals = ma_series.tolist()
ma_x = h_x[:len(ma_vals)]
# Trim leading NaN portion
while len(ma_vals) > 0 and (ma_vals[0] is None or (isinstance(ma_vals[0], float) and np.isnan(ma_vals[0]))):
ma_vals.pop(0)
ma_x.pop(0)
if ma_vals:
fig.add_trace(go.Scatter(
x=ma_x, y=[float(v) for v in ma_vals],
mode='lines', name=label,
line=dict(color=color, width=width),
opacity=0.85
))
# Prediction K-line — purple for up, yellow for down
if pred_df is not None and len(pred_df) > 0:
if len(hist_df) > 0:
last_ts = pd.Timestamp(hist_df['timestamps'].iloc[-1])
prev_ts = pd.Timestamp(hist_df['timestamps'].iloc[-2]) if len(hist_df) > 1 else last_ts - pd.Timedelta(days=1)
td = last_ts - prev_ts
if td.total_seconds() <= 0: td = pd.Timedelta(days=1)
pred_ts = [pd.Timestamp(last_ts + td * (i+1)).isoformat() for i in range(len(pred_df))]
else:
pred_ts = [str(i) for i in range(len(pred_df))]
fig.add_trace(go.Candlestick(
x=pred_ts,
open=[float(v) for v in pred_df['open'].tolist()],
high=[float(v) for v in pred_df['high'].tolist()],
low=[float(v) for v in pred_df['low'].tolist()],
close=[float(v) for v in pred_df['close'].tolist()],
name='Prediction',
increasing_line_color='#9C27B0', increasing_fillcolor='#CE93D8',
decreasing_line_color='#F9A825', decreasing_fillcolor='#FFF176'
))
# Wave overlays
if wave_analysis:
fib = wave_analysis.get('fibonacci_levels', {})
if fib:
for label, value in fib.items():
if isinstance(value, (int, float)):
fig.add_hline(y=float(value), line_dash='dash', line_color='#E040FB',
opacity=0.6, annotation_text=f'{label}: {float(value):.2f}')
sup = wave_analysis.get('support_level')
res = wave_analysis.get('resistance_level')
if isinstance(sup, (int, float)):
fig.add_hline(y=float(sup), line_dash='dot', line_color='#2196F3',
opacity=0.7, annotation_text=f'Support: {float(sup):.2f}')
if isinstance(res, (int, float)):
fig.add_hline(y=float(res), line_dash='dot', line_color='#F44336',
opacity=0.7, annotation_text=f'Resistance: {float(res):.2f}')
fig.update_layout(
title='Kronos Enhanced Prediction (Wave Theory)',
xaxis_title='Time', yaxis_title='Price',
template='plotly_white', height=600, showlegend=True,
xaxis_rangeslider_visible=False
)
return json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
def _build_main_chart(df, stock_code, wave_chart_data=None, buy_markers=None, sell_markers=None):
"""Build main candlestick chart with volume bars, wave overlays, and buy/sell markers."""
fig = go.Figure()
# Format x-axis dates
dates = [pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in df['timestamps']]
n = len(df)
low_arr = df['low'].values.astype(float)
high_arr = df['high'].values.astype(float)
# Candlestick
fig.add_trace(go.Candlestick(
x=dates, open=df['open'].tolist(), high=df['high'].tolist(),
low=df['low'].tolist(), close=df['close'].tolist(),
name='K线', increasing_line_color='#EF5350', decreasing_line_color='#26A69A'
))
# Volume bars
colors = ['#EF5350' if c >= o else '#26A69A' for c, o in zip(df['close'], df['open'])]
fig.add_trace(go.Bar(
x=dates, y=df['volume'].tolist(), name='成交量',
marker_color=colors, opacity=0.4, yaxis='y2'
))
# 20-day and 60-day moving averages
if len(df) >= 20:
ma20 = df['close'].rolling(20).mean().dropna()
fig.add_trace(go.Scatter(
x=[dates[i] for i in range(len(df)) if i in ma20.index],
y=ma20.tolist(), mode='lines', name='MA20',
line=dict(color='#FF9800', width=1)
))
if len(df) >= 60:
ma60 = df['close'].rolling(60).mean().dropna()
fig.add_trace(go.Scatter(
x=[dates[i] for i in range(len(df)) if i in ma60.index],
y=ma60.tolist(), mode='lines', name='MA60',
line=dict(color='#2196F3', width=1)
))
# Wave theory overlays: annotations & shapes
all_annotations = []
all_shapes = []
if wave_chart_data:
all_annotations.extend(wave_chart_data.get('annotations', []))
all_shapes.extend(wave_chart_data.get('shapes', []))
# Buy markers: three visual styles by signal type
if buy_markers:
for m in buy_markers:
idx = m.get('index', -1)
if 0 <= idx < n:
y_pos = low_arr[max(0, idx-1):min(n, idx+2)].min() * 0.985
label = m.get('label', '')
if '倍量关注' in label:
# 倍量关注: 粉红色圆点, 无文字
all_annotations.append({
'x': dates[idx], 'y': y_pos,
'text': '',
'showarrow': True, 'arrowhead': 0, 'arrowcolor': '#FF80AB',
'arrowsize': 1.2, 'arrowwidth': 0.1,
'font': {'size': 1, 'color': 'rgba(0,0,0,0)'},
'ax': 0, 'ay': 25,
})
elif '倍量' in label:
# 倍量买入/倍量强势: 红色三角形 + 文字
all_annotations.append({
'x': dates[idx], 'y': y_pos,
'text': '倍量买入',
'showarrow': True, 'arrowhead': 2, 'arrowcolor': '#FF1744',
'arrowsize': 1.5, 'arrowwidth': 2,
'font': {'size': 11, 'color': '#FF1744', 'family': 'Microsoft YaHei, sans-serif'},
'ax': 0, 'ay': 40,
'bordercolor': '#FF1744', 'borderwidth': 1,
'borderpad': 2,
})
else:
# E+买入/踏浪之道等: 深红色箭头 + 文字 (过滤弱信号)
score_val = m.get('score', 0)
if '待确认' in label and score_val < 15:
continue # 过滤踏浪之道低分信号
if 'E+买入' in label:
continue # E+买入信号过于频繁且参考意义不大，完全删除
display_text = label if label else '买入'
all_annotations.append({
'x': dates[idx], 'y': y_pos,
'text': display_text,
'showarrow': True, 'arrowhead': 2, 'arrowcolor': '#B71C1C',
'arrowsize': 1.5, 'arrowwidth': 2,
'font': {'size': 11, 'color': '#B71C1C', 'family': 'Microsoft YaHei, sans-serif'},
'ax': 0, 'ay': 40,
'bordercolor': '#B71C1C', 'borderwidth': 1,
'borderpad': 2,
})
# Sell markers: green triangle arrows above K-line (波段高位等, 过滤弱信号)
if sell_markers:
for m in sell_markers:
idx = m.get('index', -1)
if 0 <= idx < n:
score_val = m.get('score', 0)
label = m.get('label', '')
if 'E+卖出' in label and score_val > 45:
continue # 过滤E+卖出弱信号(只保留<45的强卖出)
y_pos = high_arr[max(0, idx-1):min(n, idx+2)].max() * 1.015
all_annotations.append({
'x': dates[idx], 'y': y_pos,
'text': label,
'showarrow': True, 'arrowhead': 2, 'arrowcolor': '#00E676',
'arrowsize': 1.5, 'arrowwidth': 2,
'font': {'size': 12, 'color': '#00E676', 'family': 'Microsoft YaHei, sans-serif'},
'ax': 0, 'ay': -40,
'bordercolor': '#00E676', 'borderwidth': 1,
'borderpad': 2,
})
if all_annotations or all_shapes:
fig.update_layout(annotations=all_annotations, shapes=all_shapes)
fig.update_layout(
title=f'{stock_code} K线走势', template='plotly_dark',
height=500, margin=dict(l=50, r=50, t=50, b=40),
xaxis_rangeslider_visible=False,
yaxis=dict(title='价格', side='right', domain=[0.25, 1]),
yaxis2=dict(title='成交量', side='left', domain=[0, 0.2], showgrid=False),
legend=dict(orientation='h', y=1.12, x=0.5, xanchor='center', font=dict(size=10)),
showlegend=True
)
return json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
def _compute_v5_signals_on_history(df):
"""基于方案E+(v5联合投票+动量过滤+市场状态自适应)在历史K线上计算信号, 生成BUY/SELL标记
返回: (buy_markers, sell_markers, v5_scores_list)
方案E+: 市场状态自适应(ADX) + v5优化权重 + v4联合投票 + 动量过滤
回测准确率: 50.4% (6只个股, 2024-01~2025-05), 自适应目标准确率>62%
"""
import numpy as np
n = len(df)
if n < 35:
return [], [], []
close = df['close'].values.astype(float)
high = df['high'].values.astype(float)
low = df['low'].values.astype(float)
volume = df['volume'].values.astype(float)
timestamps = df['timestamps'].tolist() if 'timestamps' in df.columns else list(range(n))
# 基础参数(震荡市默认)
weights = {'MA': 0.1713, 'MACD': 0.1567, 'KDJ': 0.1071,
'RSI': 0.1175, 'BOLL': 0.2939, 'VP': 0.1214, 'MOM': 0.0321}
buy_th, sell_th = 60, 40
v4_weight = 1.5
signal_boost = 1.0
mom_buy_min, mom_sell_max = 30, 70
# 市场状态自适应(只在完整数据上检测)
current_regime = 'RANGE'
try:
if n >= 60:
from market_regime import detect_market_regime, get_adaptive_params
regime_result = detect_market_regime(close, high, low, volume)
adaptive = get_adaptive_params(regime_result)
weights = dict(adaptive['weights'])
buy_th = adaptive['buy_threshold']
sell_th = adaptive['sell_threshold']
v4_weight = adaptive['v4_vote_weight']
signal_boost = adaptive['signal_boost']
mom_buy_min = adaptive['momentum_filter_buy']
mom_sell_max = adaptive['momentum_filter_sell']
current_regime = adaptive['regime']
except Exception:
pass # 自适应失败则使用默认参数
# 预计算KDJ
kk = np.full(n, 50.0); dd = np.full(n, 50.0); jj = np.full(n, 50.0)
for i in range(n):
s = max(0, i - 8)
h9 = high[s:i+1].max(); l9 = low[s:i+1].min()
rsv = (close[i] - l9) / (h9 - l9) * 100 if h9 != l9 else 50
if rsv != rsv: continue # NaN
kk[i] = 2/3 * kk[i-1] + 1/3 * rsv if i > 0 else 50
dd[i] = 2/3 * dd[i-1] + 1/3 * kk[i] if i > 0 else 50
jj[i] = 3 * kk[i] - 2 * dd[i]
# 预计算MACD(5,10,3)
ef = close[0]; es = close[0]
dif = np.zeros(n); dea = np.zeros(n)
for i in range(n):
dif[i] = ef - es
if i < n - 1:
ef = ef * 4/6 + close[i+1] * 2/6
es = es * 9/11 + close[i+1] * 2/11
for i in range(1, n):
dea[i] = dea[i-1] * 2/4 + dif[i] * 2/4
# 预计算RSI(14)
delta = np.diff(close, prepend=close[0])
gain = np.where(delta > 0, delta, 0)
loss = np.where(delta < 0, -delta, 0)
rsi = np.full(n, 50.0)
for i in range(14, n):
ag = gain[i-13:i+1].mean(); al = loss[i-13:i+1].mean()
rsi[i] = 100 if al == 0 else 100 - 100/(1+ag/al)
# 预计算BOLL(20,2)
bup = np.full(n, np.nan); bdn = np.full(n, np.nan)
for i in range(19, n):
w = close[i-19:i+1]; m = w.mean(); s = w.std()
bup[i] = m + 2*s; bdn[i] = m - 2*s
buy_markers = []
sell_markers = []
all_scores = []
for i in range(34, n): # 需要至少35根K线
# MA信号
ma5 = close[max(0,i-4):i+1].mean()
ma35 = close[max(0,i-34):i+1].mean()
ma_score = np.clip(50 + (ma5/ma35 - 1)*500, 0, 100) if ma35 > 0 else 50
# MACD信号
dn = dea[i]
macd_score = np.clip(50 + (dif[i]/dn*25 + np.sign(dif[i]-dif[max(0,i-1)])*10), 0, 100) if abs(dn) > 1e-10 else 50
# KDJ信号
dv2 = dd[i]
kdj_score = np.clip((jj[i]/dv2)*25+25, 0, 100) if abs(dv2) > 1e-10 else 50
# RSI信号
rsi_score = np.clip(rsi[i], 0, 100)
# BOLL信号 (正向计算, 后面反转)
if not np.isnan(bup[i]) and not np.isnan(bdn[i]) and bup[i] != bdn[i]:
boll_score = np.clip((close[i]-bdn[i])/(bup[i]-bdn[i])*100, 0, 100)
else:
boll_score = 50
boll_score_reversed = 100 - boll_score # 反向
# VP信号
if volume[max(0,i-1)] > 0:
vc = volume[i]/volume[max(0,i-1)]
pc = close[i]/close[max(0,i-1)] - 1 if close[max(0,i-1)] > 0 else 0
vp_score = np.clip(50 + pc*500 + (vc-1)*100, 0, 100)
else:
vp_score = 50
# MOM信号
if i >= 5:
mom_score = np.clip((close[i]/close[i-5]-1)*1000+50, 0, 100)
else:
mom_score = 50
# 方案E: v5加权评分
composite = (ma_score * weights['MA'] +
macd_score * weights['MACD'] +
kdj_score * weights['KDJ'] +
rsi_score * weights['RSI'] +
boll_score_reversed * weights['BOLL'] +
vp_score * weights['VP'] +
mom_score * weights['MOM'])
# v5信号
v5_signal = 'BUY' if composite > buy_th else ('SELL' if composite < sell_th else 'HOLD')
# v4模拟评分
v4_sim = (ma_score*0.3 + macd_score*0.3*0.3 + kdj_score*0.3*0.3 +
rsi_score*0.3*0.3 + vp_score*0.2 + mom_score*0.25 +
(boll_score_reversed + ma_score)/2*0.1 + 50*0.15)
# 联合投票 (v5权重2*boost, v4权重自适应)
votes = {'BUY': 0, 'SELL': 0, 'HOLD': 0}
if v5_signal == 'BUY': votes['BUY'] += 2.0 * signal_boost
elif v5_signal == 'SELL': votes['SELL'] += 2.0 * signal_boost
else: votes['HOLD'] += 1.0
if v4_sim > 60: votes['BUY'] += v4_weight
elif v4_sim < 40: votes['SELL'] += v4_weight
else: votes['HOLD'] += 0.5
final_signal = max(votes, key=votes.get)
# 动量过滤: 动态阈值
mom_filtered = False
if final_signal == 'BUY' and mom_score < mom_buy_min:
final_signal = 'HOLD'; mom_filtered = True
elif final_signal == 'SELL' and mom_score > mom_sell_max:
final_signal = 'HOLD'; mom_filtered = True
total_votes = sum(votes.values())
confidence = round(votes[final_signal]/total_votes*100, 1) if total_votes > 0 else 50
all_scores.append({'index': i, 'date': timestamps[i], 'score': round(composite, 1),
'close': close[i], 'signal': final_signal, 'v4_sim': round(v4_sim, 1),
'mom_score': round(mom_score, 1), 'mom_filtered': mom_filtered,
'confidence': confidence})
if final_signal == 'BUY':
label = 'E+买入' + ('(MOM过滤)' if mom_filtered else '')
buy_markers.append({
'index': int(i), 'date': timestamps[i],
'price': round(close[i], 2),
'label': label, 'score': round(composite, 1),
'confidence': confidence
})
elif final_signal == 'SELL':
label = 'E+卖出' + ('(MOM过滤)' if mom_filtered else '')
sell_markers.append({
'index': int(i), 'date': timestamps[i],
'price': round(close[i], 2),
'label': label, 'score': round(composite, 1),
'confidence': confidence
})
return buy_markers, sell_markers, all_scores
def _build_kline_chart_enhanced(df, pred_df, lookback, pred_len,
actual_df=None, start_idx=0, wave_analysis=None,
v5_buy_markers=None, v5_sell_markers=None, v5_last_score=None):
"""增强版预测K线图: 在预测区域加入v5动量信号, 改进预测斜率
pred_df的close趋势根据v5信号方向和强度进行调整
"""
if start_idx + lookback + pred_len <= len(df):
hist_df = df.iloc[start_idx:start_idx + lookback]
else:
avail = min(lookback, len(df) - start_idx)
hist_df = df.iloc[start_idx:start_idx + avail]
fig = go.Figure()
# Historical K-line
h_x = [pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in hist_df['timestamps'].tolist()]
h_close = [float(v) for v in hist_df['close'].tolist()]
fig.add_trace(go.Candlestick(
x=h_x,
open=[float(v) for v in hist_df['open'].tolist()],
high=[float(v) for v in hist_df['high'].tolist()],
low=[float(v) for v in hist_df['low'].tolist()],
close=h_close,
name='Historical',
increasing_line_color='#EF5350', decreasing_line_color='#26A69A'
))
# MA lines on historical
ma_cfg = [
(5, 'MA5', '#FF6B6B', 1.3),
(10, 'MA10', '#4ECDC4', 1.3),
(20, 'MA20', '#FFD93D', 1.3),
(60, 'MA60', '#6C5CE7', 1.3),
]
for period, label, color, width in ma_cfg:
if len(h_close) >= period:
ma_series = pd.Series(h_close).rolling(window=period, min_periods=period).mean()
ma_vals = ma_series.tolist()
ma_x = h_x[:len(ma_vals)]
while len(ma_vals) > 0 and (ma_vals[0] is None or (isinstance(ma_vals[0], float) and np.isnan(ma_vals[0]))):
ma_vals.pop(0); ma_x.pop(0)
if ma_vals:
fig.add_trace(go.Scatter(
x=ma_x, y=[float(v) for v in ma_vals],
mode='lines', name=label, line=dict(color=color, width=width), opacity=0.85
))
# v5买入/卖出标记 on historical K-line
all_annotations = []
low_arr = [float(v) for v in hist_df['low'].tolist()]
high_arr = [float(v) for v in hist_df['high'].tolist()]
if v5_buy_markers:
for m in v5_buy_markers:
idx = m.get('index', -1) - start_idx # 转为相对索引
if 0 <= idx < len(h_close):
y_pos = min(low_arr[max(0,idx-1):min(len(low_arr),idx+2)]) * 0.985
score_str = str(m.get('score', ''))
all_annotations.append({
'x': h_x[idx], 'y': y_pos,
'text': f"E+ {score_str}",
'showarrow': True, 'arrowhead': 2, 'arrowcolor': '#B71C1C',
'arrowsize': 1.5, 'arrowwidth': 2,
'font': {'size': 10, 'color': '#B71C1C', 'family': 'Microsoft YaHei'},
'ax': 0, 'ay': 35,
'bordercolor': '#B71C1C', 'borderwidth': 1, 'borderpad': 1,
})
if v5_sell_markers:
for m in v5_sell_markers:
idx = m.get('index', -1) - start_idx
if 0 <= idx < len(h_close):
y_pos = max(high_arr[max(0,idx-1):min(len(high_arr),idx+2)]) * 1.015
score_str = str(m.get('score', ''))
all_annotations.append({
'x': h_x[idx], 'y': y_pos,
'text': f"E- {score_str}",
'showarrow': True, 'arrowhead': 2, 'arrowcolor': '#00E676',
'arrowsize': 1.5, 'arrowwidth': 2,
'font': {'size': 10, 'color': '#00E676', 'family': 'Microsoft YaHei'},
'ax': 0, 'ay': -35,
'bordercolor': '#00E676', 'borderwidth': 1, 'borderpad': 1,
})
# === 预测K线增强 ===
if pred_df is not None and len(pred_df) > 0:
# 生成预测时间轴
if len(hist_df) > 0:
last_ts = pd.Timestamp(hist_df['timestamps'].iloc[-1])
prev_ts = pd.Timestamp(hist_df['timestamps'].iloc[-2]) if len(hist_df) > 1 else last_ts - pd.Timedelta(days=1)
td = last_ts - prev_ts
if td.total_seconds() <= 0: td = pd.Timedelta(days=1)
pred_ts = [pd.Timestamp(last_ts + td * (i+1)).isoformat() for i in range(len(pred_df))]
else:
pred_ts = [str(i) for i in range(len(pred_df))]
# === v5动量调整预测 ===
pred_close = [float(v) for v in pred_df['close'].tolist()]
last_hist_close = h_close[-1] if h_close else pred_close[0]
if v5_last_score is not None:
score = v5_last_score
# 根据v5评分调整预测斜率
if score > 70:
# 强看多: 放大上涨幅度
momentum_factor = 1.0 + (score - 60) / 200 # 1.05~1.2
elif score > 60:
momentum_factor = 1.0 + (score - 60) / 400 # 1.0~1.025
elif score < 30:
# 强看空: 放大下跌幅度
momentum_factor = -0.3 - (40 - score) / 200 # 趋向更负
elif score < 40:
momentum_factor = -0.1 - (40 - score) / 300
else:
momentum_factor = 0 # 中性, 保持原预测
# 应用动量调整: 以最后一根历史K线为基准, 线性插值调整
base_trend = (pred_close[-1] - last_hist_close) / max(len(pred_close), 1) if pred_close else 0
adjusted_pred = []
for j, pc in enumerate(pred_close):
progress = (j + 1) / max(len(pred_close), 1)
adjustment = momentum_factor * base_trend * progress
adjusted = last_hist_close + (pc - last_hist_close) + adjustment * last_hist_close * 0.01
adjusted_pred.append(max(adjusted, pc * 0.9)) # 不低于原预测90%
pred_close = adjusted_pred
# 重构pred_df的open/high/low以匹配调整后的close
adj_pred_open = []
adj_pred_high = []
adj_pred_low = []
for j in range(len(pred_df)):
orig_open = float(pred_df['open'].iloc[j])
orig_high = float(pred_df['high'].iloc[j])
orig_low = float(pred_df['low'].iloc[j])
ratio = pred_close[j] / max(float(pred_df['close'].iloc[j]), 1e-10)
adj_pred_open.append(orig_open * ratio)
adj_pred_high.append(orig_high * ratio)
adj_pred_low.append(orig_low * ratio)
# v5信号方向箭头 (在预测区域起始位置)
if v5_last_score is not None:
if v5_last_score > 60:
all_annotations.append({
'x': pred_ts[0], 'y': min(adj_pred_low) * 0.97,
'text': f'PlanE BUY({v5_last_score})',
'showarrow': True, 'arrowhead': 3, 'arrowcolor': '#FF1744',
'arrowsize': 2, 'arrowwidth': 2.5,
'font': {'size': 12, 'color': '#FF1744', 'family': 'Microsoft YaHei'},
'ax': 0, 'ay': 40,
})
elif v5_last_score < 40:
all_annotations.append({
'x': pred_ts[0], 'y': max(adj_pred_high) * 1.03,
'text': f'PlanE SELL({v5_last_score})',
'showarrow': True, 'arrowhead': 3, 'arrowcolor': '#00E676',
'arrowsize': 2, 'arrowwidth': 2.5,
'font': {'size': 12, 'color': '#00E676', 'family': 'Microsoft YaHei'},
'ax': 0, 'ay': -40,
})
else:
all_annotations.append({
'x': pred_ts[0], 'y': max(adj_pred_high) * 1.03,
'text': f'PlanE HOLD({v5_last_score})',
'showarrow': True, 'arrowhead': 1, 'arrowcolor': '#FFD93D',
'arrowsize': 1.5, 'arrowwidth': 2,
'font': {'size': 11, 'color': '#FFD93D', 'family': 'Microsoft YaHei'},
'ax': 0, 'ay': -35,
})
# 动量趋势线 (在预测区域画一条趋势线)
if len(pred_close) > 2:
fig.add_trace(go.Scatter(
x=[pred_ts[0], pred_ts[-1]],
y=[pred_close[0], pred_close[-1]],
mode='lines', name='PlanE趋势',
line=dict(color='#42A5F5', width=2, dash='dash'),
opacity=0.8
))
fig.add_trace(go.Candlestick(
x=pred_ts,
open=adj_pred_open, high=adj_pred_high, low=adj_pred_low, close=pred_close,
name='Prediction(PlanE)',
increasing_line_color='#9C27B0', increasing_fillcolor='#CE93D8',
decreasing_line_color='#F9A825', decreasing_fillcolor='#FFF176'
))
# Wave overlays
if wave_analysis:
fib = wave_analysis.get('fibonacci_levels', {})
if fib:
for label, value in fib.items():
if isinstance(value, (int, float)):
fig.add_hline(y=float(value), line_dash='dash', line_color='#E040FB',
opacity=0.6, annotation_text=f'{label}: {float(value):.2f}')
sup = wave_analysis.get('support_level')
res = wave_analysis.get('resistance_level')
if isinstance(sup, (int, float)):
fig.add_hline(y=float(sup), line_dash='dot', line_color='#2196F3',
opacity=0.7, annotation_text=f'Support: {float(sup):.2f}')
if isinstance(res, (int, float)):
fig.add_hline(y=float(res), line_dash='dot', line_color='#F44336',
opacity=0.7, annotation_text=f'Resistance: {float(res):.2f}')
if all_annotations:
fig.update_layout(annotations=all_annotations)
title_suffix = ''
if v5_last_score is not None:
sig = 'BUY' if v5_last_score > 60 else ('SELL' if v5_last_score < 40 else 'HOLD')
title_suffix = f' | v5: {sig}({v5_last_score})'
fig.update_layout(
title=f'Kronos Enhanced Prediction (Wave + Plan-E){title_suffix}',
xaxis_title='Time', yaxis_title='Price',
template='plotly_white', height=600, showlegend=True,
xaxis_rangeslider_visible=False
)
return json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
def _build_tech_signals(indicators_df):
"""构建完整的技术指标信号（12个固定参数指标）"""
ts = {}
def _last(col):
s = indicators_df[col].dropna()
return float(s.iloc[-1]) if len(s) > 0 else None
def _prev(col):
s = indicators_df[col].dropna()
return float(s.iloc[-2]) if len(s) >= 2 else None
# MACD(5,10,3)
m, ms, mh = _last('macd'), _last('macd_signal'), _last('macd_hist')
mp, msp = _prev('macd'), _prev('macd_signal')
if m is not None and ms is not None:
cross = None
if mp is not None and msp is not None:
if mp <= msp and m > ms: cross = 'gold'
elif mp >= msp and m < ms: cross = 'dead'
ts['macd'] = {'dif': round(m, 4), 'dea': round(ms, 4), 'hist': round(mh, 4) if mh else 0,
'signal_text': '金叉' if cross=='gold' else '死叉' if cross=='dead' else ('多头' if m > ms else '空头'),
'params': '(5,10,3)'}
# KDJ(3,3,3)
kv, dv, jv = _last('k'), _last('d'), _last('j')
kp, dp = _prev('k'), _prev('d')
if kv is not None and dv is not None:
cross = None
if kp is not None and dp is not None:
if kp <= dp and kv > dv: cross = 'gold'
elif kp >= dp and kv < dv: cross = 'dead'
over = '超买' if kv > 80 else '超卖' if kv < 20 else '正常'
ts['kdj'] = {'k': round(kv, 2), 'd': round(dv, 2), 'j': round(jv, 2) if jv else None,
'zone': over, 'cross': cross, 'params': '(3,3,3)'}
# EMA均线系统
e5, e10, e20, e60 = _last('ema_5'), _last('ema_10'), _last('ema_20'), _last('ema_60')
close_v = _last('close')
if e5 is not None and close_v is not None:
align = '多头排列' if all([e5, e10, e20, e60]) and e5 >= e10 >= e20 >= e60 else \
'空头排列' if all([e5, e10, e20, e60]) and e5 <= e10 <= e20 <= e60 else '缠绕'
ts['ema'] = {'ema5': round(e5, 2) if e5 else None, 'ema10': round(e10, 2) if e10 else None,
'ema20': round(e20, 2) if e20 else None, 'ema60': round(e60, 2) if e60 else None,
'align': align, 'params': '5/10/20/60'}
# BOLL(20,2)
bu, bm, bl = _last('bb_upper'), _last('bb_middle'), _last('bb_lower')
bp = _last('bb_position')
if bp is not None:
zone = '超买区' if bp > 0.8 else '超卖区' if bp < 0.2 else '正常区'
ts['boll'] = {'upper': round(bu, 2) if bu else None, 'middle': round(bm, 2) if bm else None,
'lower': round(bl, 2) if bl else None, 'position': round(bp, 3),
'zone': zone, 'params': '(20,2)'}
# RSI(6,14)
r6, r14 = _last('rsi_6'), _last('rsi_14')
if r14 is not None:
sig14 = '超买' if r14 > 70 else '超卖' if r14 < 30 else '中性'
ts['rsi'] = {'rsi6': round(r6, 2) if r6 else None, 'rsi14': round(r14, 2),
'signal14': sig14, 'params': '6/14'}
# CCI(20)
cc = _last('cci')
if cc is not None:
zone = '强多头' if cc > 100 else '强空头' if cc < -100 else '正常'
ts['cci'] = {'value': round(cc, 2), 'zone': zone, 'params': '(20)'}
# 成交量体系
vr5 = _last('vol_ratio5')
vol = _last('volume')
if vr5 is not None:
status = '放量' if vr5 > 1.5 else '缩量' if vr5 < 0.5 else '正常'
ts['volume'] = {'ratio5': round(vr5, 2), 'volume': int(vol) if vol else None,
'status': status, 'params': '5/10日均量'}
# OBV
obv_v = _last('obv')
obv_ma = _last('obv_ma10')
if obv_v is not None and obv_ma is not None:
ts['obv'] = {'trend': '流入' if obv_v > obv_ma else '流出', 'params': '默认'}
# DMI(14,6)
adx_v, pdi, mdi = _last('adx'), _last('plus_di'), _last('minus_di')
if adx_v is not None:
trend = '强趋势' if adx_v > 25 else '弱趋势' if adx_v > 20 else '无趋势'
direction = '多头' if pdi and mdi and pdi > mdi else '空头'
ts['dmi'] = {'adx': round(adx_v, 2), 'plus_di': round(pdi, 2) if pdi else None,
'minus_di': round(mdi, 2) if mdi else None,
'trend': trend, 'direction': direction, 'params': '(14,6)'}
# SAR(0.02,0.2)
sar_v = _last('sar')
sar_sig = _last('sar_signal')
if sar_v is not None:
ts['sar'] = {'value': round(sar_v, 2),
'signal': '多头' if sar_sig == 1 else '空头',
'params': '(0.02,0.2)'}
# MFI(14)
mfi_v = _last('mfi')
if mfi_v is not None:
zone = '资金流入' if mfi_v > 80 else '资金流出' if mfi_v < 20 else '中性'
ts['mfi'] = {'value': round(mfi_v, 2), 'zone': zone, 'params': '(14)'}
return ts
def _build_technical_charts(df, indicators_df):
"""Build technical indicator sub-charts."""
figs = {}
# RSI
if 'rsi_14' in indicators_df.columns:
rsi_vals = indicators_df['rsi_14'].dropna()
fig = go.Figure()
fig.add_trace(go.Scatter(
x=[pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in rsi_vals.index],
y=[float(v) for v in rsi_vals.tolist()],
mode='lines', name='RSI(14)', line=dict(color='#667eea', width=1.5)
))
fig.add_hline(y=70, line_dash='dash', line_color='#EF5350', opacity=0.5)
fig.add_hline(y=30, line_dash='dash', line_color='#26A69A', opacity=0.5)
fig.update_layout(title='RSI (14)', template='plotly_white', height=220,
margin=dict(l=40, r=20, t=40, b=30), showlegend=False)
figs['rsi'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
# MACD
if all(c in indicators_df.columns for c in ('macd', 'macd_signal', 'macd_hist')):
fig = go.Figure()
hist_vals = indicators_df['macd_hist'].fillna(0)
colors = ['#26A69A' if v >= 0 else '#EF5350' for v in hist_vals.tolist()]
x_axis = [pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in indicators_df.index]
fig.add_trace(go.Bar(x=x_axis, y=[float(v) for v in hist_vals.tolist()],
marker_color=colors, name='MACD Hist', opacity=0.6))
macd_vals = indicators_df['macd'].dropna()
sig_vals = indicators_df['macd_signal'].dropna()
fig.add_trace(go.Scatter(x=[pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in macd_vals.index],
y=[float(v) for v in macd_vals.tolist()],
mode='lines', name='MACD', line=dict(color='#667eea', width=1.5)))
fig.add_trace(go.Scatter(x=[pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in sig_vals.index],
y=[float(v) for v in sig_vals.tolist()],
mode='lines', name='Signal', line=dict(color='#FF7043', width=1.5)))
fig.update_layout(title='MACD', template='plotly_white', height=220,
margin=dict(l=40, r=20, t=40, b=30), showlegend=True)
figs['macd'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
# Bollinger Bands
if all(c in indicators_df.columns for c in ('bb_upper', 'bb_middle', 'bb_lower')):
fig = go.Figure()
x_axis = [pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in indicators_df.index]
fig.add_trace(go.Scatter(x=x_axis, y=[float(v) for v in indicators_df['bb_upper'].tolist()],
mode='lines', name='Upper', line=dict(width=1)))
fig.add_trace(go.Scatter(x=x_axis, y=[float(v) for v in indicators_df['bb_lower'].tolist()],
mode='lines', name='Lower', line=dict(width=1), fill='tonexty',
fillcolor='rgba(102,126,234,0.1)'))
fig.add_trace(go.Scatter(x=x_axis, y=[float(v) for v in indicators_df['bb_middle'].tolist()],
mode='lines', name='Middle', line=dict(color='#667eea', width=1.5)))
if 'close' in df.columns:
fig.add_trace(go.Scatter(x=x_axis, y=[float(v) for v in df['close'].tolist()],
mode='lines', name='Close', line=dict(color='black', width=1)))
fig.update_layout(title='Bollinger Bands', template='plotly_white', height=220,
margin=dict(l=40, r=20, t=40, b=30), showlegend=True)
figs['bollinger'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
# Volume
if 'volume' in df.columns:
fig = go.Figure()
x_axis = [pd.Timestamp(x).isoformat() if not isinstance(x, str) else x for x in df.index]
colors = ['#26A69A' if float(c) >= float(o) else '#EF5350'
for c, o in zip(df['close'].tolist(), df['open'].tolist())]
fig.add_trace(go.Bar(x=x_axis, y=[float(v) for v in df['volume'].tolist()],
marker_color=colors, opacity=0.7))
fig.update_layout(title='Volume', template='plotly_white', height=180,
margin=dict(l=40, r=20, t=40, b=30), showlegend=False)
figs['volume'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
return figs
if 'rsi_14' in indicators_df.columns:
fig = go.Figure()
fig.add_trace(go.Scatter(
x=indicators_df.index, y=indicators_df['rsi_14'],
mode='lines', name='RSI(14)', line=dict(color='#667eea', width=1.5)
))
fig.add_hline(y=70, line_dash='dash', line_color='#EF5350', opacity=0.5)
fig.add_hline(y=30, line_dash='dash', line_color='#26A69A', opacity=0.5)
fig.add_hrect(y0=30, y1=70, fillcolor='gray', opacity=0.05)
fig.update_layout(title='RSI (14)', template='plotly_white', height=250,
margin=dict(l=40, r=20, t=40, b=30), showlegend=False)
figs['rsi'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
if all(c in indicators_df.columns for c in ('macd', 'macd_signal', 'macd_hist')):
fig = go.Figure()
colors = ['#26A69A' if v >= 0 else '#EF5350' for v in indicators_df['macd_hist'].fillna(0)]
fig.add_trace(go.Bar(x=indicators_df.index, y=indicators_df['macd_hist'],
marker_color=colors, name='MACD Hist', opacity=0.6))
fig.add_trace(go.Scatter(x=indicators_df.index, y=indicators_df['macd'],
mode='lines', name='MACD', line=dict(color='#667eea', width=1.5)))
fig.add_trace(go.Scatter(x=indicators_df.index, y=indicators_df['macd_signal'],
mode='lines', name='Signal', line=dict(color='#FF7043', width=1.5)))
fig.update_layout(title='MACD', template='plotly_white', height=250,
margin=dict(l=40, r=20, t=40, b=30), showlegend=True)
figs['macd'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
if all(c in indicators_df.columns for c in ('bb_upper', 'bb_middle', 'bb_lower')):
fig = go.Figure()
fig.add_trace(go.Scatter(x=indicators_df.index, y=indicators_df['bb_upper'],
mode='lines', name='Upper', line=dict(width=1)))
fig.add_trace(go.Scatter(x=indicators_df.index, y=indicators_df['bb_lower'],
mode='lines', name='Lower', line=dict(width=1),
fill='tonexty', fillcolor='rgba(102,126,234,0.1)'))
fig.add_trace(go.Scatter(x=indicators_df.index, y=indicators_df['bb_middle'],
mode='lines', name='Middle', line=dict(color='#667eea', width=1.5)))
if 'close' in df.columns:
fig.add_trace(go.Scatter(x=df.index, y=df['close'],
mode='lines', name='Close', line=dict(color='black', width=1)))
fig.update_layout(title='Bollinger Bands', template='plotly_white', height=250,
margin=dict(l=40, r=20, t=40, b=30), showlegend=True)
figs['bollinger'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
if 'volume' in df.columns:
fig = go.Figure()
colors = ['#26A69A' if c >= o else '#EF5350'
for c, o in zip(df['close'], df['open'])]
fig.add_trace(go.Bar(x=df.index, y=df['volume'], marker_color=colors, opacity=0.7))
fig.update_layout(title='Volume', template='plotly_white', height=200,
margin=dict(l=40, r=20, t=40, b=30), showlegend=False)
figs['volume'] = json.dumps(fig, cls=plotly.utils.PlotlyJSONEncoder)
return figs
# ============================================================
# Routes
# ============================================================
@app.route('/api/wave-analysis', methods=['POST'])
def wave_analysis_endpoint():
"""Dedicated wave analysis endpoint supporting multi-period (day/m60/m30)."""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
period = data.get('period', 'day') # 'day', 'm60', 'm30'
if not stock_code:
return jsonify({'error': 'Stock code is required'}), 400
# Download data according to period
if period == 'day':
df, err = _download_stock_data(stock_code)
else:
df, err = _download_period_data(stock_code, period=period)
if err:
return jsonify({'error': err}), 400
if len(df) < 30:
return jsonify({'error': f'Insufficient data: only {len(df)} bars'}), 400
lookback = int(data.get('lookback', 0))
if lookback > 0 and len(df) > lookback:
df = df.iloc[-lookback:].reset_index(drop=True)
result = {}
if WAVE_ANALYSIS_AVAILABLE:
try:
df_wv = df.set_index('timestamps') if 'timestamps' in df.columns else df
wave_result = analyze_wave_patterns(df_wv)
result = wave_result.get('analysis', wave_result)
# Build chart with wave overlays
wave_chart = {
'annotations': wave_result.get('annotations', []),
'shapes': wave_result.get('shapes', [])
}
# v5信号标记 (主图)
_cp_buy, _cp_sell = [], []
try:
v5_buy_m, v5_sell_m, v5_scores = _compute_v5_signals_on_history(df)
_cp_buy.extend(v5_buy_m)
_cp_sell.extend(v5_sell_m)
except Exception as _v5e:
pass
chart = _build_main_chart(df, f"{stock_code}_{period}", wave_chart_data=wave_chart,
buy_markers=_cp_buy, sell_markers=_cp_sell)
except Exception as _wae:
result = {'error': str(_wae)}
chart = None
else:
result = {'error': 'wave_analysis module not loaded'}
chart = None
period_labels = {'day': '日线', 'm60': '60分钟', 'm30': '30分钟'}
info = {
'stock_code': stock_code,
'period': period,
'period_label': period_labels.get(period, period),
'bars': len(df),
'start_date': str(df['timestamps'].min())[:10],
'end_date': str(df['timestamps'].max())[:10]
}
return jsonify({
'success': True,
'data_info': info,
'wave_analysis': result,
'chart': chart
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/download-stock', methods=['POST'])
def download_stock():
"""Download stock data via AkShare by stock code."""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': 'Stock code is required'}), 400
if not stock_code.isdigit() or len(stock_code) not in (5, 6):
return jsonify({'error': 'Invalid stock code format, expected 5 or 6 digits'}), 400
start_date = data.get('start_date', '20200101')
lookback = int(data.get('lookback', 800))
df, err = _download_stock_data(stock_code, start_date=start_date)
if err:
return jsonify({'error': err}), 400
# Limit to last N rows for performance; lookback=0 means use all data
if lookback > 0 and len(df) > lookback:
df = df.iloc[-lookback:].reset_index(drop=True)
info = {
'rows': len(df),
'stock_code': stock_code,
'lookback': lookback,
'start_date': df['timestamps'].min().isoformat(),
'end_date': df['timestamps'].max().isoformat(),
'price_range': {
'min': float(df[['open','high','low','close']].min().min()),
'max': float(df[['open','high','low','close']].max().max())
},
'last_close': float(df['close'].iloc[-1]),
'enhanced_available': ENHANCED_AVAILABLE
}
# Run analysis if enhanced modules available
analysis = {}
if ENHANCED_AVAILABLE:
try:
df_analysis = df.set_index('timestamps').copy()
if 'volume' not in df_analysis.columns:
df_analysis['volume'] = 0
indicators_df = technical_analyzer.calculate_all_indicators(df_analysis.copy())
tech_signals = _build_tech_signals(indicators_df)
vp_result = volume_price_analyzer.analyze_volume_price_relationship(df_analysis)
vp_clean = {k: round(float(v), 4) if isinstance(v, (np.floating, float)) else v
for k, v in vp_result.items()}
wave_result = wave_analyzer.analyze_wave_patterns(df_analysis)
wave_clean = {}
for k, v in wave_result.items():
if isinstance(v, dict):
wave_clean[k] = {kk: round(float(vv), 4) if isinstance(vv, (np.floating, float, np.integer, int)) else vv
for kk, vv in v.items()}
elif isinstance(v, (np.floating, float, np.integer, int)):
wave_clean[k] = round(float(v), 4)
elif isinstance(v, np.ndarray):
wave_clean[k] = [round(float(x), 4) for x in v.tolist()]
else:
wave_clean[k] = v
tech_charts = _build_technical_charts(df_analysis, indicators_df)
analysis = {
'technical_signals': tech_signals,
'volume_price_analysis': vp_clean,
'wave_analysis': wave_clean,
'charts': tech_charts
}
# Classic pattern detection
if CLASSIC_PATTERN_AVAILABLE:
try:
cp_result = detect_classic_patterns(df_analysis)
analysis['classic_patterns'] = {
'summary': cp_result.get('summary', ''),
'score': cp_result.get('score', 0),
'overall_signal': cp_result.get('overall_signal', 'none'),
'detected_list': cp_result.get('detected_list', []),
}
except Exception as _cpe:
analysis['classic_patterns'] = {'error': str(_cpe)}
except Exception as e:
print(f"[ENHANCED ERROR] {type(e).__name__}: {e}")
analysis = {'error': str(e)}
# Classic pattern detection (independent of enhanced modules)
cp_buy_markers = []
cp_sell_markers = []
if CLASSIC_PATTERN_AVAILABLE:
try:
df_cp = df.set_index('timestamps') if 'timestamps' in df.columns else df
cp_result = detect_classic_patterns(df_cp)
cp_buy_markers = cp_result.get('buy_markers', [])
cp_sell_markers = cp_result.get('sell_markers', [])
# Extract all double_volume signals from _all_signals for chart markers
patterns = cp_result.get('patterns', {})
dv_pattern = patterns.get('double_volume', {})
if dv_pattern.get('_all_signals'):
for sig in dv_pattern['_all_signals']:
idx = sig.get('index', -1)
if idx >= 0:
label = '倍量强势' if sig.get('signal') == 'bullish' else '倍量关注'
cp_buy_markers.append({
'index': int(idx),
'date': sig.get('date', ''),
'price': sig.get('price', 0),
'label': label,
'strength': sig.get('strength', 0),
})
# Extract all distribution_top signals for chart markers
dt_pattern = patterns.get('distribution_top', {})
if dt_pattern.get('_all_signals'):
for sig in dt_pattern['_all_signals']:
idx = sig.get('index', -1)
if idx >= 0:
cp_sell_markers.append({
'index': int(idx),
'date': sig.get('date', ''),
'price': sig.get('price', 0),
'label': sig.get('label', '波段高位'),
'strength': sig.get('strength', 0),
})
if isinstance(analysis, dict) and 'error' not in analysis:
analysis['classic_patterns'] = {
'summary': cp_result.get('summary', ''),
'score': cp_result.get('score', 0),
'overall_signal': cp_result.get('overall_signal', 'none'),
'detected_list': cp_result.get('detected_list', []),
}
else:
if not isinstance(analysis, dict):
analysis = {}
analysis['classic_patterns'] = {
'summary': cp_result.get('summary', ''),
'score': cp_result.get('score', 0),
'overall_signal': cp_result.get('overall_signal', 'none'),
'detected_list': cp_result.get('detected_list', []),
}
except Exception as _cpe:
if isinstance(analysis, dict):
analysis['classic_patterns'] = {'error': str(_cpe)}
else:
analysis = {'classic_patterns': {'error': str(_cpe)}}
# Wyckoff VPA analysis (independent of enhanced modules)
if WYCKOFF_VPA_AVAILABLE:
try:
df_wv = df.set_index('timestamps') if 'timestamps' in df.columns else df
wv_result = detect_wyckoff_vpa(df_wv)
if isinstance(analysis, dict):
analysis['wyckoff_vpa'] = wv_result
except Exception as _wve:
analysis['wyckoff_vpa'] = {'error': str(_wve)}
# Wave Theory analysis (independent of enhanced modules)
wave_chart_data = None
if WAVE_ANALYSIS_AVAILABLE:
try:
df_wv = df.set_index('timestamps') if 'timestamps' in df.columns else df
wave_result = analyze_wave_patterns(df_wv)
# wave_result = {'annotations':..., 'shapes':..., 'analysis':{waves, buy_signals, sell_signals, ...}}
if isinstance(analysis, dict):
analysis['wave_analysis'] = wave_result.get('analysis', wave_result)
wave_chart_data = {
'annotations': wave_result.get('annotations', []),
'shapes': wave_result.get('shapes', [])
}
except Exception as _wae:
analysis['wave_analysis'] = {'error': str(_wae)}
# 踏浪之道 · 主升浪评分分析
if TALANG_AVAILABLE:
try:
df_tl = df.copy()
df_tl['timestamps'] = df_tl.get('timestamps', df_tl.index)
talang_result = talang_score_analysis(df_tl, stock_code)
if talang_result.get('success'):
# 序列化处理: 将numpy类型转为Python原生类型
def _talang_serialize(obj):
if isinstance(obj, dict):
return {k: _talang_serialize(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [_talang_serialize(v) for v in obj]
elif isinstance(obj, (np.bool_,)):
return bool(obj)
elif isinstance(obj, (np.integer,)):
return int(obj)
elif isinstance(obj, (np.floating,)):
return float(obj)
elif isinstance(obj, np.ndarray):
return obj.tolist()
return obj
talang_result = _talang_serialize(talang_result)
analysis['talang'] = talang_result
# 将踏浪之道的买卖标注合并到经典形态标记中
for m in talang_result.get('buy_markers', []):
cp_buy_markers.append(m)
except Exception as _tle:
analysis['talang'] = {'error': str(_tle)}
# Extract HK stock name if applicable
hk_name = ''
if len(stock_code) == 5:
try:
import urllib.request as _urllib
_url2 = f'https://web.ifzq.gtimg.cn/appstock/app/hkfqkline/get?param=hk{stock_code},day,,,1,qfq'
_req2 = _urllib.Request(_url2, headers={'User-Agent': 'Mozilla/5.0'})
_resp2 = _urllib.urlopen(_req2, timeout=10)
_raw2 = _resp2.read().decode()
if _raw2.startswith('kline_dayqfq'):
_raw2 = _raw2[_raw2.index('(')+1:_raw2.rindex(')')]
_jd2 = json.loads(_raw2)
_qt2 = _jd2.get('data', {}).get(f'hk{stock_code}', {}).get('qt', {}).get(f'hk{stock_code}', [])
if isinstance(_qt2, list) and len(_qt2) > 1:
hk_name = str(_qt2[1])
except Exception:
pass
# Build candlestick chart for the frontend
try:
# v5信号标记 (download-stock主图)
try:
v5_buy_m2, v5_sell_m2, _ = _compute_v5_signals_on_history(df)
cp_buy_markers.extend(v5_buy_m2)
cp_sell_markers.extend(v5_sell_m2)
except Exception:
pass
chart = _build_main_chart(df, stock_code, wave_chart_data=wave_chart_data,
buy_markers=cp_buy_markers, sell_markers=cp_sell_markers)
except Exception as _ce:
chart = None
return jsonify({'success': True, 'data_info': info, 'analysis': analysis, 'hk_name': hk_name, 'chart': chart})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/')
def index():
resp = make_response(render_template('index.html'))
resp.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
resp.headers['Pragma'] = 'no-cache'
resp.headers['Expires'] = '0'
return resp
@app.route('/api/data-files')
def get_data_files():
return jsonify(_load_data_files())
@app.route('/api/load-data', methods=['POST'])
def load_data():
try:
data = request.get_json()
fp = data.get('file_path')
if not fp:
return jsonify({'error': 'File path is required'}), 400
df, err = _load_csv_file(fp)
if err:
return jsonify({'error': err}), 400
def detect_timeframe(df_):
if len(df_) < 2:
return 'Unknown'
diffs = [(df_['timestamps'].iloc[i] - df_['timestamps'].iloc[i-1])
for i in range(1, min(10, len(df_)))]
avg = sum(diffs, pd.Timedelta(0)) / len(diffs)
if avg < pd.Timedelta(minutes=1):
return f"{avg.total_seconds():.0f} seconds"
elif avg < pd.Timedelta(hours=1):
return f"{avg.total_seconds()/60:.0f} minutes"
elif avg < pd.Timedelta(days=1):
return f"{avg.total_seconds()/3600:.0f} hours"
else:
return f"{avg.days} days"
info = {
'rows': len(df),
'columns': list(df.columns),
'start_date': df['timestamps'].min().isoformat(),
'end_date': df['timestamps'].max().isoformat(),
'price_range': {
'min': float(df[['open','high','low','close']].min().min()),
'max': float(df[['open','high','low','close']].max().max())
},
'timeframe': detect_timeframe(df),
'enhanced_available': ENHANCED_AVAILABLE
}
return jsonify({'success': True, 'data_info': info})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/available-models')
def get_available_models():
return jsonify({
'models': AVAILABLE_MODELS,
'model_available': MODEL_AVAILABLE,
'enhanced_available': ENHANCED_AVAILABLE
})
@app.route('/api/model-status')
def get_model_status():
loaded = predictor is not None or enhanced_predictor is not None
return jsonify({
'available': MODEL_AVAILABLE,
'loaded': loaded,
'enhanced_available': ENHANCED_AVAILABLE,
'message': 'Model loaded' if loaded else 'No model loaded'
})
@app.route('/api/load-model', methods=['POST'])
def load_model():
global tokenizer, model, predictor, enhanced_predictor
try:
if not MODEL_AVAILABLE:
return jsonify({'error': 'Kronos model library not available'}), 400
data = request.get_json()
model_key = data.get('model_key', 'kronos-small')
device = data.get('device', 'cpu')
use_enhanced = data.get('use_enhanced', False)
if model_key not in AVAILABLE_MODELS:
return jsonify({'error': f'Unsupported model: {model_key}'}), 400
cfg = AVAILABLE_MODELS[model_key]
tokenizer = KronosTokenizer.from_pretrained(cfg['tokenizer_id'])
model = Kronos.from_pretrained(cfg['model_id'])
predictor = KronosPredictor(model, tokenizer, device=device, max_context=cfg['context_length'])
if use_enhanced and ENHANCED_AVAILABLE:
try:
enhanced_predictor = EnhancedKronosPredictor(model, tokenizer, device=device,
max_context=cfg['context_length'])
except Exception:
enhanced_predictor = None
else:
enhanced_predictor = None
return jsonify({
'success': True,
'message': f'Model loaded: {cfg["name"]} ({cfg["params"]}) on {device}',
'model_info': {
'name': cfg['name'], 'params': cfg['params'],
'context_length': cfg['context_length'],
'enhanced': enhanced_predictor is not None
}
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/enhanced-analysis', methods=['POST'])
def enhanced_analysis():
"""Run technical indicators + wave theory + volume-price analysis."""
if not ENHANCED_AVAILABLE:
return jsonify({'error': 'Enhanced modules not available'}), 400
try:
data = request.get_json()
fp = data.get('file_path')
if not fp:
return jsonify({'error': 'File path is required'}), 400
df, err = _load_csv_file_for_analysis(fp)
if err:
return jsonify({'error': err}), 400
# 1. Technical indicators
indicators_df = technical_analyzer.calculate_all_indicators(df.copy())
tech_signals = _build_tech_signals(indicators_df)
# 2. Volume-price
vp_result = volume_price_analyzer.analyze_volume_price_relationship(df)
vp_clean = {k: round(float(v), 4) if isinstance(v, (np.floating, float)) else v
for k, v in vp_result.items()}
# 3. Wave theory
wave_result = wave_analyzer.analyze_wave_patterns(df)
wave_clean = {}
for k, v in wave_result.items():
if isinstance(v, dict):
wave_clean[k] = {kk: round(float(vv), 4) if isinstance(vv, (np.floating, float, np.integer, int)) else vv
for kk, vv in v.items()}
elif isinstance(v, (np.floating, float, np.integer, int)):
wave_clean[k] = round(float(v), 4)
elif isinstance(v, np.ndarray):
wave_clean[k] = [round(float(x), 4) for x in v.tolist()]
else:
wave_clean[k] = v
# 4. Charts
tech_charts = _build_technical_charts(df, indicators_df)
return jsonify({
'success': True,
'technical_signals': tech_signals,
'volume_price': vp_clean,
'wave_theory': wave_clean,
'charts': tech_charts,
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/multi-dimension-analysis', methods=['POST'])
def multi_dimension_analysis():
"""多维度技术指标联动研判：12指标 + 5维度 + 信号共振"""
if not ENHANCED_AVAILABLE or multidim_analyzer is None:
return jsonify({'error': 'Enhanced modules not available'}), 400
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
lookback = int(data.get('lookback', 250))
# 通过 _download_stock_data 获取日线数据
df, err = _download_stock_data(stock_code)
if err or df is None:
return jsonify({'error': f'获取数据失败: {err}'}), 400
# 按 lookback 截取
if lookback > 0 and len(df) > lookback:
df = df.tail(lookback).reset_index(drop=True)
if len(df) < 60:
return jsonify({'error': '数据不足，至少需要60根K线'}), 400
# 执行多维度研判
result = multidim_analyzer.analyze(df)
return jsonify({
'success': True,
'multi_dimension': result,
})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
@app.route('/api/enhanced-predict', methods=['POST'])
def enhanced_predict():
"""Run prediction using enhanced predictor with wave theory."""
try:
data = request.get_json()
fp = data.get('file_path')
if not fp:
return jsonify({'error': 'File path is required'}), 400
df, err = _load_csv_file(fp)
if err:
return jsonify({'error': err}), 400
lookback = int(data.get('lookback', 400))
pred_len = int(data.get('pred_len', 120))
temperature = float(data.get('temperature', 1.0))
top_p = float(data.get('top_p', 0.9))
sample_count = int(data.get('sample_count', 1))
use_enhanced = data.get('use_enhanced', True)
if len(df) < lookback:
return jsonify({'error': f'Need at least {lookback} rows, got {len(df)}'}), 400
start_date = data.get('start_date')
if start_date:
start_dt = pd.to_datetime(start_date)
mask = df['timestamps'] >= start_dt
window_df = df[mask]
if len(window_df) < lookback + pred_len:
return jsonify({'error': 'Insufficient data in selected window'}), 400
x_df = window_df.iloc[:lookback]
actual_df = window_df.iloc[lookback:lookback+pred_len]
start_idx = df[mask].index[0] if len(df[mask]) > 0 else 0
else:
x_df = df.iloc[:lookback]
actual_df = df.iloc[lookback:lookback+pred_len] if len(df) >= lookback + pred_len else None
start_idx = 0
required_cols = ['open', 'high', 'low', 'close']
if 'volume' in x_df.columns:
required_cols.append('volume')
x_input = x_df[required_cols].copy()
x_timestamp = pd.Series(x_df['timestamps'].values)
y_timestamp = x_timestamp
# Run prediction
wave_analysis = None
if ENHANCED_AVAILABLE and use_enhanced and enhanced_predictor is not None:
try:
pred_df = enhanced_predictor.predict_enhanced(
x_input, x_timestamp, y_timestamp, pred_len,
T=temperature, top_p=top_p, sample_count=sample_count, verbose=False
)
except Exception:
if predictor is not None:
pred_df = predictor.predict(x_input, x_timestamp, y_timestamp, pred_len,
T=temperature, top_p=top_p, sample_count=sample_count)
else:
return jsonify({'error': 'No model available'}), 400
elif predictor is not None:
pred_df = predictor.predict(x_input, x_timestamp, y_timestamp, pred_len,
T=temperature, top_p=top_p, sample_count=sample_count)
else:
return jsonify({'error': 'No model loaded, please load a model first'}), 400
# Wave analysis on historical data
if ENHANCED_AVAILABLE and use_enhanced:
try:
hist_for_wave = x_input.copy()
hist_for_wave['timestamps'] = x_timestamp.values
hist_for_wave = hist_for_wave.set_index('timestamps')
if 'volume' not in hist_for_wave.columns:
hist_for_wave['volume'] = 0
wave_raw = wave_analyzer.analyze_wave_patterns(hist_for_wave)
wave_analysis = {}
for k, v in wave_raw.items():
if isinstance(v, dict):
wave_analysis[k] = {kk: round(float(vv), 4) if isinstance(vv, (np.floating, float, np.integer, int)) else vv
for kk, vv in v.items()}
elif isinstance(v, (np.floating, float, np.integer, int)):
wave_analysis[k] = round(float(v), 4)
else:
wave_analysis[k] = v
except Exception as e:
wave_analysis = {'error': str(e)}
# v5动量增强预测图
v5_buy_m3 = []; v5_sell_m3 = []; v5_last = None
try:
df_for_v5 = df.set_index('timestamps') if 'timestamps' in df.columns else df
if V5_SCORER_AVAILABLE:
v5r = _v5_mod.run_v5_optimized_analysis(stock_code, df_for_v5)
v5_last = v5r.get('score', None)
v5_buy_m3, v5_sell_m3, _ = _compute_v5_signals_on_history(df)
except Exception:
pass
chart_json = _build_kline_chart_enhanced(df, pred_df, lookback, pred_len,
actual_df, start_idx, wave_analysis,
v5_buy_markers=v5_buy_m3, v5_sell_markers=v5_sell_m3,
v5_last_score=v5_last)
pred_results = []
for _, row in pred_df.iterrows():
pred_results.append({
'open': float(row['open']), 'high': float(row['high']),
'low': float(row['low']), 'close': float(row['close']),
'volume': float(row['volume']) if 'volume' in row else 0
})
actual_data = []
if actual_df is not None:
for _, row in actual_df.iterrows():
actual_data.append({
'open': float(row['open']), 'high': float(row['high']),
'low': float(row['low']), 'close': float(row['close']),
'volume': float(row['volume']) if 'volume' in row else 0,
'timestamp': row['timestamps'].isoformat() if 'timestamps' in row else ''
})
return jsonify({
'success': True,
'chart': chart_json,
'prediction': pred_results,
'actual': actual_data,
'wave_analysis': wave_analysis,
'enhanced': use_enhanced and ENHANCED_AVAILABLE,
'has_comparison': len(actual_data) > 0
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/fundamental-analysis', methods=['POST'])
def fundamental_analysis_route():
"""基本面分析API - AI Hedge Fund风格多Agent分析"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
# 清理股票代码
stock_code = stock_code.replace('.SH', '').replace('.SZ', '').replace('.sh', '').replace('.sz', '')
if not FUNDAMENTAL_AVAILABLE:
return jsonify({'error': '基本面分析模块未加载'}), 503
result = run_fundamental_analysis(stock_code)
return jsonify(result)
except Exception as e:
return jsonify({'error': f'分析失败: {str(e)}'}), 500
@app.route('/api/joint-prediction', methods=['POST'])
def joint_prediction():
"""联合预测: 方案E+ - v5优化权重 + v4联合投票 + 动量过滤 + 市场状态自适应
回测准确率50.4%, 6只个股, 2024-01~2025-05
E+: ADX趋势检测 + 动态阈值/权重/信号增强 + v5信号权重2 + v4权重自适应
"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
results = {}
# 0. 市场状态检测
regime_result = None
adaptive_params = {}
try:
df_regime, _ = _download_stock_data(stock_code)
if df_regime is not None and len(df_regime) >= 60:
from market_regime import detect_market_regime, get_adaptive_params as _gap
regime_result = detect_market_regime(
df_regime['close'].values.astype(float),
df_regime['high'].values.astype(float),
df_regime['low'].values.astype(float),
df_regime['volume'].values.astype(float)
)
adaptive_params = _gap(regime_result)
results['market_regime'] = {
'regime': regime_result['regime'],
'regime_label': regime_result['regime_label'],
'adx': regime_result['adx'],
'di_plus': regime_result['di_plus'],
'di_minus': regime_result['di_minus'],
'atr_pct': regime_result['atr_pct'],
'volatility': regime_result['volatility'],
'volume_state': regime_result['volume_state'],
'streak': regime_result['streak'],
'price_vs_ma60': regime_result['price_vs_ma60'],
}
except Exception as e:
results['regime_error'] = str(e)[:100]
# 1. v5优化评分
if V5_SCORER_AVAILABLE:
try:
df_v5, err_v5 = _download_stock_data(stock_code)
if not err_v5 and df_v5 is not None:
if 'timestamps' in df_v5.columns:
df_v5 = df_v5.set_index('timestamps')
v5_result = _v5_mod.run_v5_optimized_analysis(stock_code, df_v5)
results['v5_score'] = v5_result['score']
results['v5_signal'] = v5_result.get('signal', 'HOLD')
results['v5_details'] = v5_result.get('v5_details', {})
except Exception as e:
results['v5_error'] = str(e)[:100]
# 2. 方案E+联合投票 + 动量过滤 + 市场状态自适应
try:
if V5_SCORER_AVAILABLE:
v5_sc = results.get('v5_score', 50)
v5_sig = results.get('v5_signal', 'HOLD')
mom_val = results.get('v5_details', {}).get('details', {}).get('MOM', {}).get('raw_score', 50)
v4_sc = results.get('v4_score', 50)
# 自适应参数
a_boost = adaptive_params.get('signal_boost', 1.0) if adaptive_params else 1.0
a_v4w = adaptive_params.get('v4_vote_weight', 1.5) if adaptive_params else 1.5
a_mom_buy = adaptive_params.get('momentum_filter_buy', 30) if adaptive_params else 30
a_mom_sell = adaptive_params.get('momentum_filter_sell', 70) if adaptive_params else 70
# 联合投票 (v5权重2*boost, v4权重自适应)
votes_e = {'BUY': 0, 'SELL': 0, 'HOLD': 0}
if v5_sig == 'BUY': votes_e['BUY'] += 2.0 * a_boost
elif v5_sig == 'SELL': votes_e['SELL'] += 2.0 * a_boost
else: votes_e['HOLD'] += 1.0
if v4_sc > 60: votes_e['BUY'] += a_v4w
elif v4_sc < 40: votes_e['SELL'] += a_v4w
else: votes_e['HOLD'] += 0.5
e_signal = max(votes_e, key=votes_e.get)
total_v = sum(votes_e.values())
e_conf = round(votes_e[e_signal]/total_v*100, 1) if total_v > 0 else 50
# 动量过滤(动态阈值)
mom_filtered = False
if e_signal == 'BUY' and mom_val < a_mom_buy:
e_signal = 'HOLD'; mom_filtered = True
elif e_signal == 'SELL' and mom_val > a_mom_sell:
e_signal = 'HOLD'; mom_filtered = True
trend_map = {'BUY': '看涨', 'SELL': '看跌', 'HOLD': '震荡'}
regime_label = results.get('market_regime', {}).get('regime_label', '')
plan_note = f'方案E+: v5联合投票+动量过滤+市场状态自适应(50.4%回测准确率)'
if regime_label:
plan_note += f' [{regime_label}]'
results['plan_e'] = {
'signal': e_signal,
'trend_direction': trend_map.get(e_signal, '震荡'),
'confidence': e_conf,
'v5_score': v5_sc,
'v5_signal': v5_sig,
'v4_score': v4_sc,
'v4_weight': a_v4w,
'mom_score': mom_val,
'mom_filtered': mom_filtered,
'signal_boost': a_boost,
'votes': {k: round(v, 1) for k, v in votes_e.items()},
'source': 'plan_e_plus',
'regime': results.get('market_regime', {}),
'note': plan_note,
}
results['ft_prediction'] = results['plan_e']
# E+共振信息
results['ft_resonance'] = {
'agreement': 1 if e_signal in ('BUY', 'SELL') else 0,
'strength': e_conf / 100,
'final_direction': trend_map.get(e_signal, '震荡'),
'v5_signal': v5_sig,
'plan_e_signal': e_signal,
}
except Exception as e:
results['ft_error'] = str(e)[:100]
# 3. 方案F-v5: MACD(5,10,3)+高置信模式(BUY>70/SELL<28)+6因子方向修正 (准确率55.5%)
try:
if V53_SCORER_AVAILABLE:
df_fv3, err_fv3 = _download_stock_data(stock_code)
if not err_fv3 and df_fv3 is not None:
if 'timestamps' in df_fv3.columns:
df_fv3 = df_fv3.set_index('timestamps')
fv3_result = _v53_mod.run_fv3_analysis(stock_code, df_fv3)
fv4_signal = fv3_result.get('signal', 'HOLD')
fv4_score = fv3_result.get('score', 50)
fv3_conf = fv3_result.get('confidence', 50)
fv4_tier = fv3_result.get('tier', 'HOLD')
results['fv5_score'] = fv4_score
results['fv5_signal'] = fv4_signal
results['fv5_confidence'] = fv3_conf
results['fv5_tier'] = fv4_tier
results['fv5_label'] = fv3_result.get('label', '')
results['plan_fv3'] = {
'signal': fv4_signal,
'trend_direction': {'BUY': '看涨', 'SELL': '看跌', 'HOLD': '震荡'}.get(fv4_signal, '震荡'),
'confidence': fv3_conf,
'score': fv4_score,
'tier': fv4_tier,
'source': 'plan_f_v3',
'note': '方案F-v5: MACD(5,10,3)+高置信模式+方向修正(MACD+KDJ+VWAP正向,MA+VP+RSI反向) 准确率55.5%',
'label': fv3_result.get('label', ''),
}
# F-v3作为主预测(替代E+)
results['ft_prediction'] = results['plan_fv3']
results['ft_resonance']['plan_fv4_signal'] = fv4_signal
except Exception as e:
results['fv4_error'] = str(e)[:200]
# 4. v4五维评分 (如果可用)
if V4_SCORER_AVAILABLE:
try:
df_v4, err_v4 = _download_stock_data(stock_code)
if not err_v4 and df_v4 is not None:
if 'timestamps' in df_v4.columns:
df_v4 = df_v4.set_index('timestamps')
v4_result = _v4_mod.run_v4_composite_analysis(stock_code, df_v4)
results['v4_score'] = v4_result.get('composite_score', 0)
results['v4_trend'] = v4_result.get('trend_direction', '')
except Exception as e:
results['v4_error'] = str(e)[:100]
# 4. 综合研判
v5_signal = results.get('v5_signal', 'HOLD')
ft_trend = results.get('plan_e', {}).get('trend_direction', '震荡')
ft_res = results.get('ft_resonance', {})
ft_agreement = ft_res.get('agreement', 0)
ft_strength = ft_res.get('strength', 0)
# 方案E投票 (v5信号 + plan_e信号)
votes = {'BUY': 0, 'SELL': 0, 'HOLD': 0}
# v5信号权重2
if v5_signal == 'BUY': votes['BUY'] += 2
elif v5_signal == 'SELL': votes['SELL'] += 2
else: votes['HOLD'] += 1
# 方案E信号权重1.5 (联合投票+动量过滤结果)
plan_e_sig = results.get('plan_e', {}).get('signal', 'HOLD')
if plan_e_sig == 'BUY': votes['BUY'] += 1.5
elif plan_e_sig == 'SELL': votes['SELL'] += 1.5
else: votes['HOLD'] += 0.5
total_votes = sum(votes.values())
if total_votes > 0:
final_signal = max(votes, key=votes.get)
confidence = votes[final_signal] / total_votes * 100
else:
final_signal = 'HOLD'
confidence = 50
results['joint_prediction'] = {
'final_signal': final_signal,
'confidence': round(confidence, 1),
'votes': {k: round(v, 1) for k, v in votes.items()},
'systems_used': {
'v5_optimized': 'v5_score' in results,
'ft_transformer': 'plan_e' in results,
'v4_composite': 'v4_score' in results,
'ft_resonance': ft_agreement >= 2,
}
}
return jsonify({'success': True, 'stock_code': stock_code, 'data': results})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': f'联合预测失败: {str(e)}'}), 500
# ============================================================
# v4 五维综合评分 API
# ============================================================
try:
_v4_scorer_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'v4_composite_scorer.py')
import importlib.util as _ilu_v4
_v4_spec = _ilu_v4.spec_from_file_location('v4_composite_scorer', _v4_scorer_path)
_v4_mod = _ilu_v4.module_from_spec(_v4_spec)
_v4_spec.loader.exec_module(_v4_mod)
V4_SCORER_AVAILABLE = True
print("[INFO] v4五维评分模块已加载")
except Exception as e:
V4_SCORER_AVAILABLE = False
print("[WARN] v4五维评分模块加载失败: {}".format(e))
@app.route('/api/v4-composite-analysis', methods=['POST'])
def v4_composite_analysis():
"""v4五维综合评分: 基本面+动量+浪型+技术+资金流 + 市场状态自适应"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
if not V4_SCORER_AVAILABLE:
return jsonify({'error': 'v4评分模块未加载'}), 503
# 获取数据 - 使用已有的_download_stock_data
df, err = _download_stock_data(stock_code)
if err:
return jsonify({'error': err}), 400
if 'timestamps' in df.columns:
df = df.set_index('timestamps') if df.index.name != 'timestamps' else df
# 确保必要的列存在
required = ['open', 'high', 'low', 'close', 'volume']
for col in required:
if col not in df.columns:
return jsonify({'error': f'缺少必要列: {col}'}), 400
# 执行五维分析
result = _v4_mod.run_v4_composite_analysis(stock_code, df)
return jsonify({'success': True, 'data': result})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': f'分析失败: {str(e)}'}), 500
# v5 优化评分模块 (基于回测验证最优权重)
# ============================================================
try:
_v5_scorer_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'v5_optimized_scorer.py')
import importlib.util as _ilu_v5
_v5_spec = _ilu_v5.spec_from_file_location('v5_optimized_scorer', _v5_scorer_path)
_v5_mod = _ilu_v5.module_from_spec(_v5_spec)
_v5_spec.loader.exec_module(_v5_mod)
V5_SCORER_AVAILABLE = True
print("[INFO] v5优化评分模块已加载 (方案E+: 联合+动量过滤+市场状态自适应, 准确率: 50.4%)")
except Exception as e:
V5_SCORER_AVAILABLE = False
print("[WARN] v5优化评分模块加载失败: {}".format(e))
# v5.3 增强评分模块 (方案F-v5: MACD(5,10,3)+高置信模式(BUY>70/SELL<28)+6因子方向修正, 准确率55.5%)
# ============================================================
try:
_v53_scorer_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'v5_enhanced_scorer.py')
import importlib.util as _ilu_v53
_v53_spec = _ilu_v53.spec_from_file_location('v5_enhanced_scorer', _v53_scorer_path)
_v53_mod = _ilu_v53.module_from_spec(_v53_spec)
_v53_spec.loader.exec_module(_v53_mod)
V53_SCORER_AVAILABLE = True
print("[INFO] v5.3增强评分模块已加载 (方案F-v5: MACD(5,10,3)+高置信模式(BUY>70/SELL<28)+6因子方向修正, 准确率: 55.5%)")
except Exception as e:
V53_SCORER_AVAILABLE = False
# 蒸鱼ML预测引擎 (GBDT日线模型, 测试集54.0%, 阈值0.5%, 16股600日)
# ============================================================
try:
_ml_engine_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'ml_engine.py')
import importlib.util as _ilu_ml
_ml_spec = _ilu_ml.spec_from_file_location('ml_engine', _ml_engine_path)
_ml_mod = _ilu_ml.module_from_spec(_ml_spec)
_ml_spec.loader.exec_module(_ml_mod)
ML_ENGINE_AVAILABLE = True
print("[INFO] 蒸鱼ML引擎已加载 (GBDT 54.0%, 16股重新训练)")
except Exception as e:
ML_ENGINE_AVAILABLE = False
print("[WARN] 蒸鱼ML引擎加载失败: {}".format(e))
print("[WARN] v5.3增强评分模块加载失败: {}".format(e))
@app.route('/api/v5-optimized-analysis', methods=['POST'])
def v5_optimized_analysis():
"""v5优化评分: 基于回测验证最优权重的7维指标综合评分
回测准确率: 54.42% (884有效信号/4398样本)
配置: KDJ(3,3,3) MACD(5,10,3) MA(5,17,26,35,105,155,250)
布林带反向, 其余正向; BUY>60, SELL<40
"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
if not V5_SCORER_AVAILABLE:
return jsonify({'error': 'v5评分模块未加载'}), 503
df, err = _download_stock_data(stock_code)
if err:
return jsonify({'error': err}), 400
if 'timestamps' in df.columns:
df = df.set_index('timestamps') if df.index.name != 'timestamps' else df
required = ['open', 'high', 'low', 'close', 'volume']
for col in required:
if col not in df.columns:
return jsonify({'error': f'缺少必要列: {col}'}), 400
result = _v5_mod.run_v5_optimized_analysis(stock_code, df)
return jsonify({'success': True, 'data': result})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': f'分析失败: {str(e)}'}), 500
@app.route('/api/predict', methods=['POST'])
def predict():
"""Legacy prediction endpoint (backward compatible)."""
try:
data = request.get_json()
fp = data.get('file_path')
lookback = int(data.get('lookback', 400))
pred_len = int(data.get('pred_len', 120))
temperature = float(data.get('temperature', 1.0))
top_p = float(data.get('top_p', 0.9))
sample_count = int(data.get('sample_count', 1))
if not fp:
return jsonify({'error': 'File path is required'}), 400
df, err = _load_csv_file(fp)
if err:
return jsonify({'error': err}), 400
if predictor is None:
return jsonify({'error': 'Model not loaded'}), 400
x_df = df.iloc[:lookback][['open','high','low','close']].copy()
if 'volume' in df.columns:
x_df['volume'] = df.iloc[:lookback]['volume']
x_ts = pd.Series(df.iloc[:lookback]['timestamps'].values)
y_ts = pd.Series(df.iloc[lookback:lookback+pred_len]['timestamps'].values)
pred_df = predictor.predict(x_df, x_ts, y_ts, pred_len,
T=temperature, top_p=top_p, sample_count=sample_count)
actual_data = []
actual_df = None
if len(df) >= lookback + pred_len:
actual_df = df.iloc[lookback:lookback+pred_len]
for _, row in actual_df.iterrows():
actual_data.append({
'open': float(row['open']), 'high': float(row['high']),
'low': float(row['low']), 'close': float(row['close']),
'volume': float(row['volume']) if 'volume' in row else 0
})
# v5动量增强 (offline预测)
v5_last2 = None
try:
df_for_v5_2 = df.set_index('timestamps') if 'timestamps' in df.columns else df
if V5_SCORER_AVAILABLE:
v5r2 = _v5_mod.run_v5_optimized_analysis('', df_for_v5_2)
v5_last2 = v5r2.get('score', None)
except Exception:
pass
chart_json = _build_kline_chart_enhanced(df, pred_df, lookback, pred_len, actual_df, v5_last_score=v5_last2)
pred_results = []
for _, row in pred_df.iterrows():
pred_results.append({
'open': float(row['open']), 'high': float(row['high']),
'low': float(row['low']), 'close': float(row['close']),
'volume': float(row['volume']) if 'volume' in row else 0
})
return jsonify({
'success': True, 'chart': chart_json,
'prediction_results': pred_results, 'actual_data': actual_data,
'has_comparison': len(actual_data) > 0
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/quick-predict', methods=['POST'])
def quick_predict():
"""Full pipeline: download data -> analyze -> predict, all by stock code."""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': 'Stock code is required'}), 400
lookback = int(data.get('lookback', 400))
pred_len = int(data.get('pred_len', 120))
temperature = float(data.get('temperature', 1.0))
top_p = float(data.get('top_p', 0.9))
use_enhanced = data.get('use_enhanced', True)
# Step 1: Download data
df, err = _download_stock_data(stock_code)
if err:
return jsonify({'error': err}), 400
# lookback=0 means use all available data
if lookback == 0:
lookback = len(df)
if len(df) < lookback:
lookback = len(df) - 1
if lookback < 50:
return jsonify({'error': f'Insufficient data: only {len(df)} rows'}), 400
# Step 2: Analysis
analysis = {}
if ENHANCED_AVAILABLE:
try:
df_analysis = df.set_index('timestamps').copy()
if 'volume' not in df_analysis.columns:
df_analysis['volume'] = 0
indicators_df = technical_analyzer.calculate_all_indicators(df_analysis.copy())
tech_signals = _build_tech_signals(indicators_df)
vp_result = volume_price_analyzer.analyze_volume_price_relationship(df_analysis)
vp_clean = {k: round(float(v), 4) if isinstance(v, (np.floating, float)) else v
for k, v in vp_result.items()}
wave_result = wave_analyzer.analyze_wave_patterns(df_analysis)
wave_clean = {}
for k, v in wave_result.items():
if isinstance(v, dict):
wave_clean[k] = {kk: round(float(vv), 4) if isinstance(vv, (np.floating, float, np.integer, int)) else vv
for kk, vv in v.items()}
elif isinstance(v, (np.floating, float, np.integer, int)):
wave_clean[k] = round(float(v), 4)
elif isinstance(v, np.ndarray):
wave_clean[k] = [round(float(x), 4) for x in v.tolist()]
else:
wave_clean[k] = v
tech_charts = _build_technical_charts(df_analysis, indicators_df)
analysis = {
'technical_signals': tech_signals,
'volume_price_analysis': vp_clean,
'wave_analysis': wave_clean,
'charts': tech_charts
}
# Classic pattern detection
if CLASSIC_PATTERN_AVAILABLE:
try:
cp_result = detect_classic_patterns(df_analysis)
analysis['classic_patterns'] = {
'summary': cp_result.get('summary', ''),
'score': cp_result.get('score', 0),
'overall_signal': cp_result.get('overall_signal', 'none'),
'detected_list': cp_result.get('detected_list', []),
}
except Exception as _cpe:
analysis['classic_patterns'] = {'error': str(_cpe)}
except Exception as e:
analysis = {'error': str(e)}
# Classic pattern detection (independent of enhanced modules)
cp_buy_markers = []
cp_sell_markers = []
if CLASSIC_PATTERN_AVAILABLE:
try:
df_cp = df.set_index('timestamps') if 'timestamps' in df.columns else df
cp_result = detect_classic_patterns(df_cp)
cp_buy_markers = cp_result.get('buy_markers', [])
cp_sell_markers = cp_result.get('sell_markers', [])
# Extract all double_volume signals from _all_signals for chart markers
patterns = cp_result.get('patterns', {})
dv_pattern = patterns.get('double_volume', {})
if dv_pattern.get('_all_signals'):
for sig in dv_pattern['_all_signals']:
idx = sig.get('index', -1)
if idx >= 0:
label = '倍量强势' if sig.get('signal') == 'bullish' else '倍量关注'
cp_buy_markers.append({
'index': int(idx),
'date': sig.get('date', ''),
'price': sig.get('price', 0),
'label': label,
'strength': sig.get('strength', 0),
})
# Extract all distribution_top signals for chart markers
dt_pattern = patterns.get('distribution_top', {})
if dt_pattern.get('_all_signals'):
for sig in dt_pattern['_all_signals']:
idx = sig.get('index', -1)
if idx >= 0:
cp_sell_markers.append({
'index': int(idx),
'date': sig.get('date', ''),
'price': sig.get('price', 0),
'label': sig.get('label', '波段高位'),
'strength': sig.get('strength', 0),
})
if isinstance(analysis, dict) and 'error' not in analysis:
analysis['classic_patterns'] = {
'summary': cp_result.get('summary', ''),
'score': cp_result.get('score', 0),
'overall_signal': cp_result.get('overall_signal', 'none'),
'detected_list': cp_result.get('detected_list', []),
}
else:
if not isinstance(analysis, dict):
analysis = {}
analysis['classic_patterns'] = {
'summary': cp_result.get('summary', ''),
'score': cp_result.get('score', 0),
'overall_signal': cp_result.get('overall_signal', 'none'),
'detected_list': cp_result.get('detected_list', []),
}
except Exception as _cpe:
if isinstance(analysis, dict):
analysis['classic_patterns'] = {'error': str(_cpe)}
else:
analysis = {'classic_patterns': {'error': str(_cpe)}}
# Wyckoff VPA analysis (independent of enhanced modules)
if WYCKOFF_VPA_AVAILABLE:
try:
df_wv = df.set_index('timestamps') if 'timestamps' in df.columns else df
wv_result = detect_wyckoff_vpa(df_wv)
if isinstance(analysis, dict):
analysis['wyckoff_vpa'] = wv_result
except Exception as _wve:
analysis['wyckoff_vpa'] = {'error': str(_wve)}
# Wave Theory analysis (independent of enhanced modules)
if WAVE_ANALYSIS_AVAILABLE:
try:
df_wv = df.set_index('timestamps') if 'timestamps' in df.columns else df
wave_result = analyze_wave_patterns(df_wv)
if isinstance(analysis, dict):
analysis['wave_analysis'] = wave_result.get('analysis', wave_result)
except Exception as _wae:
analysis['wave_analysis'] = {'error': str(_wae)}
# 踏浪之道 · 主升浪评分分析
if TALANG_AVAILABLE:
try:
df_tl = df.copy()
df_tl['timestamps'] = df_tl.get('timestamps', df_tl.index)
talang_result = talang_score_analysis(df_tl, stock_code)
if talang_result.get('success'):
analysis['talang'] = talang_result
# 将踏浪之道的买卖标注合并到经典形态标记中
for m in talang_result.get('buy_markers', []):
cp_buy_markers.append(m)
except Exception as _tle:
analysis['talang'] = {'error': str(_tle)}
# Step 3: Prediction
x_df = df.iloc[-lookback:].reset_index(drop=True)
required_cols = ['open', 'high', 'low', 'close']
if 'volume' in x_df.columns:
required_cols.append('volume')
x_input = x_df[required_cols].copy()
x_timestamp = pd.Series(x_df['timestamps'].values)
wave_analysis_for_chart = analysis.get('wave_analysis') if analysis else None
if ENHANCED_AVAILABLE and use_enhanced and enhanced_predictor is not None:
try:
pred_df = enhanced_predictor.predict_enhanced(
x_input, x_timestamp, x_timestamp, pred_len,
T=temperature, top_p=top_p, verbose=False
)
except Exception:
if predictor is not None:
pred_df = predictor.predict(x_input, x_timestamp, x_timestamp, pred_len,
T=temperature, top_p=top_p)
else:
# Offline fallback
pred_df = _offline_predict(x_input, pred_len)
elif predictor is not None:
pred_df = predictor.predict(x_input, x_timestamp, x_timestamp, pred_len,
T=temperature, top_p=top_p)
else:
pred_df = _offline_predict(x_input, pred_len)
# v5动量增强 (wave分析预测)
v5_last3 = None; v5_bm3 = []; v5_sm3 = []
try:
if V5_SCORER_AVAILABLE:
x_for_v5 = x_df.set_index('timestamps') if 'timestamps' in x_df.columns else x_df
v5r3 = _v5_mod.run_v5_optimized_analysis(stock_code, x_for_v5)
v5_last3 = v5r3.get('score', None)
v5_bm3, v5_sm3, _ = _compute_v5_signals_on_history(x_df)
except Exception:
pass
chart_json = _build_kline_chart_enhanced(x_df, pred_df, lookback, pred_len, wave_analysis=wave_analysis_for_chart,
v5_buy_markers=v5_bm3, v5_sell_markers=v5_sm3,
v5_last_score=v5_last3)
pred_results = []
for _, row in pred_df.iterrows():
pred_results.append({
'open': float(row['open']), 'high': float(row['high']),
'low': float(row['low']), 'close': float(row['close']),
'volume': float(row['volume']) if 'volume' in row else 0
})
info = {
'rows': len(x_df),
'stock_code': stock_code,
'lookback': lookback,
'start_date': x_df['timestamps'].min().isoformat(),
'end_date': x_df['timestamps'].max().isoformat(),
'price_range': {
'min': float(x_df[['open','high','low','close']].min().min()),
'max': float(x_df[['open','high','low','close']].max().max())
},
'last_close': float(x_df['close'].iloc[-1]),
'pred_start': pred_results[0]['close'] if pred_results else 0,
'pred_end': pred_results[-1]['close'] if pred_results else 0,
'pred_change_pct': round((pred_results[-1]['close'] / pred_results[0]['close'] - 1) * 100, 2) if pred_results and pred_results[0]['close'] > 0 else 0
}
# Extract HK stock name if applicable
hk_name = ''
is_hk = len(stock_code) == 5
if is_hk:
try:
import urllib.request as _urllib
_url = f'https://web.ifzq.gtimg.cn/appstock/app/hkfqkline/get?param=hk{stock_code},day,,,1,qfq'
_req = _urllib.Request(_url, headers={'User-Agent': 'Mozilla/5.0'})
_resp = _urllib.urlopen(_req, timeout=10)
_raw = _resp.read().decode()
if _raw.startswith('kline_dayqfq'):
_raw = _raw[_raw.index('(')+1:_raw.rindex(')')]
_jd = json.loads(_raw)
qt = _jd.get('data', {}).get(f'hk{stock_code}', {}).get('qt', {}).get(f'hk{stock_code}', [])
if isinstance(qt, list) and len(qt) > 1:
hk_name = str(qt[1])
except Exception:
pass
return jsonify({
'success': True,
'data_info': info,
'analysis': analysis,
'chart': chart_json,
'prediction': pred_results,
'enhanced': use_enhanced and ENHANCED_AVAILABLE,
'hk_name': hk_name
})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
def _offline_predict(x_input, pred_len):
"""Fallback predictor using simple statistics when no model is loaded."""
last_close = float(x_input['close'].iloc[-1])
last_vol = float(x_input['volume'].iloc[-1]) if 'volume' in x_input.columns else 0
# Calculate recent volatility
returns = x_input['close'].pct_change().dropna()
daily_vol = float(returns.std()) if len(returns) > 1 else 0.015
daily_drift = float(returns.mean()) if len(returns) > 1 else 0.0002
predictions = []
import random
np.random.seed(42)
for i in range(pred_len):
change = daily_drift + daily_vol * np.random.randn()
new_close = last_close * (1 + change)
new_high = new_close * (1 + abs(daily_vol * np.random.randn()) * 0.5)
new_low = new_close * (1 - abs(daily_vol * np.random.randn()) * 0.5)
new_open = (last_close + new_close) / 2 * (1 + np.random.randn() * 0.002)
new_vol = last_vol * (0.8 + np.random.rand() * 0.4)
predictions.append({
'open': round(max(new_open, new_low), 2),
'high': round(max(new_high, new_open, new_close), 2),
'low': round(min(new_low, new_open, new_close), 2),
'close': round(new_close, 2),
'volume': round(new_vol, 2)
})
last_close = new_close
return pd.DataFrame(predictions)
# ============================================================
# 分析报告导出 API
# ============================================================
# ====== 踏浪之道 · 主升浪分析 API ======
@app.route('/api/talang-analysis', methods=['POST'])
def talang_analysis_endpoint():
"""踏浪之道主升浪评分分析 - 基于已下载的数据进行15条规则评分"""
if not TALANG_AVAILABLE:
return jsonify({'error': '踏浪之道模块未加载'}), 503
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
min_score = int(data.get('min_score', 8))
# 使用缓存的数据或重新下载
cache_key = f"{stock_code}_2024-01-01"
if cache_key in stock_data_cache:
df = stock_data_cache[cache_key]
else:
from talang_integration import fetch_sina_data, normalize_df, calc_indicators
if stock_code.startswith('6'):
sina_code = f'sh{stock_code}'
else:
sina_code = f'sz{stock_code}'
df = fetch_sina_data(sina_code, 600)
if df is None or len(df) < 120:
return jsonify({'error': f'无法获取{stock_code}数据或数据不足'}), 400
df = df[df['date'] >= '2024-01-01'].reset_index(drop=True)
stock_data_cache[cache_key] = df
# 转换为Kronos格式的DataFrame
df_kronos = df.copy()
df_kronos['timestamps'] = df_kronos['date']
result = talang_score_analysis(df_kronos, stock_code, min_score=min_score)
return jsonify(result)
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
@app.route('/api/talang-backtest', methods=['POST'])
def talang_backtest_endpoint():
"""踏浪之道完整回测 - 下载历史数据执行15条规则回测"""
if not TALANG_AVAILABLE:
return jsonify({'error': '踏浪之道模块未加载'}), 503
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
name = data.get('name', '')
min_score = int(data.get('min_score', 8))
result = talang_backtest_full(stock_code, name=name, min_score=min_score)
if 'error' in result:
return jsonify(result), 400
return jsonify(result)
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
@app.route('/api/export-report', methods=['POST'])
def export_report_route():
"""导出完整分析报告为独立HTML"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
stock_code = stock_code.replace('.SH', '').replace('.SZ', '').replace('.sh', '').replace('.sz', '')
lookback = int(data.get('lookback', 250))
report = {'stock_code': stock_code, 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
# 1. K线+技术分析(复用quick-predict逻辑, 不做预测)
try:
df, err = _download_stock_data(stock_code)
if not err and df is not None:
info = {
'rows': len(df),
'start_date': str(df['timestamps'].min())[:10],
'end_date': str(df['timestamps'].max())[:10],
'last_close': float(df['close'].iloc[-1])
}
report['kline_info'] = info
# 技术分析
if ENHANCED_AVAILABLE:
try:
df_a = df.set_index('timestamps').copy()
if 'volume' not in df_a.columns:
df_a['volume'] = 0
indicators_df = technical_analyzer.calculate_all_indicators(df_a.copy())
tech_signals = _build_tech_signals(indicators_df)
report['tech_signals'] = tech_signals
except Exception as te:
report['tech_error'] = str(te)[:200]
# 威科夫
if WYCKOFF_VPA_AVAILABLE:
try:
df_wv = df.set_index('timestamps') if 'timestamps' in df.columns else df
wv_result = detect_wyckoff_vpa(df_wv)
report['wyckoff'] = wv_result
except Exception:
pass
# 经典形态
if CLASSIC_PATTERN_AVAILABLE:
try:
df_cp = df.set_index('timestamps') if 'timestamps' in df.columns else df
cp_result = detect_classic_patterns(df_cp)
report['classic_patterns'] = cp_result
except Exception:
pass
# 波浪
if WAVE_ANALYSIS_AVAILABLE:
try:
df_wv = df.set_index('timestamps') if 'timestamps' in df.columns else df
wave_result = analyze_wave_patterns(df_wv)
report['wave'] = wave_result
except Exception:
pass
except Exception as e:
report['kline_error'] = str(e)[:200]
# 2. 基本面分析
if FUNDAMENTAL_AVAILABLE:
try:
fa_result = run_fundamental_analysis(stock_code)
report['fundamental'] = fa_result
except Exception as e:
report['fundamental_error'] = str(e)[:200]
# 4. v4五维评分
if V4_SCORER_AVAILABLE:
try:
df_v4, err_v4 = _download_stock_data(stock_code)
if not err_v4 and df_v4 is not None:
if 'timestamps' in df_v4.columns:
df_v4 = df_v4.set_index('timestamps')
v4_result = _v4_mod.run_v4_composite_analysis(stock_code, df_v4)
report['v4_composite'] = v4_result
except Exception as e:
report['v4_error'] = str(e)[:200]
# 4. 多维度联动
try:
md_result = multi_dimension_analysis(stock_code)
report['multi_dimension'] = md_result
except Exception as e:
report['multi_dim_error'] = str(e)[:200]
# 生成HTML报告
html = _build_export_html(report)
fname = 'Kronos_{}_{}.html'.format(stock_code, datetime.now().strftime('%Y%m%d_%H%M%S'))
return jsonify({'success': True, 'html': html, 'filename': fname})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': '导出失败: {}'.format(str(e))}), 500
def _build_export_html(report):
"""构建独立可离线查看的HTML报告"""
code = report.get('stock_code', 'unknown')
ts = report.get('timestamp', '')
# 股票信息
si = {}
fa = report.get('fundamental', {})
if fa and fa.get('agents'):
first_agent = fa['agents'][0] or {}
si = first_agent.get('stock_info', {})
name = si.get('name', code)
price = si.get('price', 0)
pe = si.get('pe', 0)
ki = report.get('kline_info', {})
last_close = ki.get('last_close', price)
# 基本面共识
cs = fa.get('consensus') or {}
signal = cs.get('signal', 'neutral')
fa_score = cs.get('final_score', 0)
summary = cs.get('summary', '')
sig_color = {'bullish': '#e74c3c', 'bearish': '#26a69a', 'neutral': '#f0ad4e'}.get(signal, '#999')
sig_label = {'bullish': '看多', 'bearish': '看空', 'neutral': '中性'}.get(signal, '未知')
# v4五维
v4 = report.get('v4_composite', {})
v4_score = v4.get('composite_score', 0)
v4_trend = v4.get('trend_direction', '')
v4_dims = v4.get('dimensions', [])
v4_thresholds = v4.get('thresholds', {})
# 技术信号
tech = report.get('tech_signals', {})
# 威科夫
wv = report.get('wyckoff', {})
# agents
agents = fa.get('agents', [])
# 经典形态
cp = report.get('classic_patterns', {})
cp_summary = cp.get('summary', '') if isinstance(cp, dict) else ''
cp_score = cp.get('score', 0) if isinstance(cp, dict) else 0
# 构建HTML
lines = []
lines.append('<!DOCTYPE html><html lang="zh-CN"><head><meta charset="UTF-8">')
lines.append('<meta name="viewport" content="width=device-width,initial-scale=1.0">')
lines.append('<title>Kronos Report - {} ({})</title>'.format(name, code))
lines.append('<style>')
lines.append('*{margin:0;padding:0;box-sizing:border-box}')
lines.append('body{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif;background:#0d1117;color:#c9d1d9;max-width:900px;margin:0 auto;padding:20px}')
lines.append('.hdr{text-align:center;padding:24px 0 16px;border-bottom:1px solid #21262d;margin-bottom:20px}')
lines.append('.hdr h1{font-size:22px;color:#f0f6fc;margin-bottom:6px}')
lines.append('.hdr .meta{font-size:12px;color:#484f58}')
lines.append('.sc{display:flex;gap:16px;margin-bottom:20px}')
lines.append('.sb{flex:1;padding:16px;border-radius:10px;text-align:center;background:rgba(255,255,255,0.03);border:1px solid #21262d}')
lines.append('.sb .n{font-size:32px;font-weight:700}')
lines.append('.sb .l{font-size:11px;color:#7a8a9e;margin-top:4px}')
lines.append('.sec{margin-bottom:16px;padding:14px;background:rgba(255,255,255,0.02);border-radius:8px;border:1px solid #21262d}')
lines.append('.sec h2{font-size:15px;color:#f0f6fc;margin-bottom:10px;padding-bottom:6px;border-bottom:1px solid #21262d}')
lines.append('.ac{padding:10px;margin:6px 0;border-radius:6px;border-left:3px solid;background:rgba(255,255,255,0.02)}')
lines.append('.ac .ah{display:flex;justify-content:space-between;align-items:center;margin-bottom:4px}')
lines.append('.ac .an{font-size:13px;font-weight:600}')
lines.append('.tg{font-size:10px;padding:2px 8px;border-radius:4px;color:#fff}')
lines.append('.tg-b{background:#e74c3c}.tg-s{background:#26a69a}.tg-n{background:#f0ad4e;color:#000}')
lines.append('.dt{display:flex;flex-wrap:wrap;gap:4px;margin-top:4px}')
lines.append('.dt span{font-size:10px;padding:2px 6px;border-radius:3px;background:rgba(255,255,255,0.05);color:#8b949e}')
lines.append('.db{display:flex;align-items:center;gap:8px;margin:5px 0;font-size:12px}')
lines.append('.db .bb{flex:1;height:14px;background:rgba(255,255,255,0.04);border-radius:3px;overflow:hidden}')
lines.append('.db .bf{height:100%;border-radius:3px}')
lines.append('.db .sv{min-width:36px;text-align:right;font-weight:600;font-size:12px}')
lines.append('.ft{text-align:center;padding:16px;font-size:10px;color:#484f58;border-top:1px solid #21262d;margin-top:20px}')
lines.append('.tag-row{display:flex;gap:6px;flex-wrap:wrap;margin-top:8px}')
lines.append('.tag-item{font-size:10px;padding:3px 8px;border-radius:4px;background:rgba(255,255,255,0.04);color:#8b949e}')
lines.append('</style></head><body>')
lines.append('<div class="hdr"><h1>{} ({})</h1>'.format(name, code))
lines.append('<div class="meta">Kronos Enhanced Web UI | {} | Price: {} | PE: {} | Data: {} rows</div>'.format(
ts, last_close, pe, ki.get('rows', '-')))
lines.append('</div>')
# 评分卡
lines.append('<div class="sc">')
lines.append('<div class="sb"><div class="n" style="color:{}">{}</div><div class="l">基本面共识 {}</div><div class="l">{}</div></div>'.format(sig_color, fa_score, sig_label, summary))
lines.append('<div class="sb"><div class="n" style="color:#42A5F5">{}</div><div class="l">五维综合评分 {}</div></div>'.format(round(v4_score, 1) if v4_score else '-', v4_trend or '-'))
if cp_score:
cp_sig = cp.get('overall_signal', 'neutral') if isinstance(cp, dict) else 'neutral'
cp_c = {'bullish': '#e74c3c', 'bearish': '#26a69a', 'neutral': '#f0ad4e'}.get(cp_sig, '#999')
lines.append('<div class="sb"><div class="n" style="color:{}">{}</div><div class="l">K线形态</div><div class="l">{}</div></div>'.format(cp_c, cp_score, cp_summary[:30] if cp_summary else ''))
lines.append('</div>')
# 五维评分
if v4_dims:
lines.append('<div class="sec"><h2>五维评分详情</h2>')
dim_colors = {'基本面': '#66BB6A', '动量': '#42A5F5', '浪型': '#AB47BC', '技术': '#FFA726', '资金流': '#EF5350'}
for dim in v4_dims:
dn = dim.get('name', '')
ds = dim.get('score', 0)
dc = dim_colors.get(dn, '#999')
pct = max(0, min(100, ds))
lines.append('<div class="db"><span style="min-width:46px">{}</span><div class="bb"><div class="bf" style="width:{}%;background:{}"></div></div><span class="sv" style="color:{}">{}</span><span style="font-size:10px;color:#666">{}%</span></div>'.format(dn, pct, dc, dc, round(ds, 1), round(dim.get('weight', 0)*100)))
lines.append('</div>')
# 基本面Agents
if agents:
lines.append('<div class="sec"><h2>基本面Agent分析</h2>')
for agent in agents:
if not agent.get('available'):
continue
a_s = agent.get('signal', 'neutral')
a_c = {'bullish': '#e74c3c', 'bearish': '#26a69a', 'neutral': '#f0ad4e'}.get(a_s, '#999')
a_l = {'bullish': '看多', 'bearish': '看空', 'neutral': '中性'}.get(a_s, '未知')
a_cls = {'bullish': 'tg-b', 'bearish': 'tg-s', 'neutral': 'tg-n'}.get(a_s, 'tg-n')
lines.append('<div class="ac" style="border-color:{}">'.format(a_c))
lines.append('<div class="ah"><span class="an">{} {}</span><span class="tg {}">{} {}/{}</span></div>'.format(agent.get('agent_icon', ''), agent.get('agent_name', ''), a_cls, a_l, agent.get('overall_score', 0), agent.get('max_score', 0)))
details = agent.get('details', [])
if details:
lines.append('<div class="dt">')
for d in details:
lines.append('<span>{}</span>'.format(d))
lines.append('</div>')
if agent.get('intrinsic_value'):
lines.append('<div style="margin-top:4px;font-size:11px">内在价值: <strong style="color:#66bb6a">{}</strong>元</div>'.format(round(agent['intrinsic_value'], 2)))
lines.append('</div>')
lines.append('</div>')
# 技术信号
if tech:
lines.append('<div class="sec"><h2>技术指标</h2>')
if isinstance(tech, dict):
for k, v in tech.items():
if isinstance(v, dict):
sig = v.get('signal', '')
sc = v.get('score', 0)
c = '#e74c3c' if sig == 'bullish' else '#26a69a' if sig == 'bearish' else '#f0ad4e'
lines.append('<div class="tag-item" style="color:{}">{}</div>'.format(c, '{}: {} ({})'.format(k, sig, sc)))
lines.append('</div>')
# 威科夫
if wv and isinstance(wv, dict):
phase = wv.get('phase', '')
if phase:
lines.append('<div class="sec"><h2>威科夫分析</h2>')
lines.append('<div class="tag-item">阶段: {}</div>'.format(phase))
events = wv.get('events', [])
for ev in events[:5]:
if isinstance(ev, dict):
lines.append('<div class="tag-item">{}: {}</div>'.format(ev.get('type', ''), ev.get('description', '')[:60]))
lines.append('</div>')
# Footer
lines.append('<div class="ft">Kronos Enhanced Web UI - AI Hedge Fund Agent Architecture<br>数据来源: 腾讯财经HTTP API | 威科夫量价分析 | 波浪理论 | 经典K线形态</div>')
lines.append('</body></html>')
return '\n'.join(lines)
if __name__ == '__main__':
print("=" * 50)
print("Kronos Enhanced Web UI")
print("=" * 50)
print(f"Model available: {MODEL_AVAILABLE}")
print(f"Enhanced modules: {ENHANCED_AVAILABLE}")
# ============================================================
# 多因子机器学习预测引擎路由 (v2.0)
# ============================================================
try:
_mf_routes_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'output', 'multifactor_routes.py')
import importlib.util as _ilu_mf
_mf_spec = _ilu_mf.spec_from_file_location('multifactor_routes_v3', _mf_routes_path)
_mf_mod = _ilu_mf.module_from_spec(_mf_spec)
_mf_spec.loader.exec_module(_mf_mod)
_mf_mod.register_routes(app)
print("[INFO] 多因子预测引擎路由已加载 v3 (6个端点)")
except Exception as e:
print("[WARN] 多因子预测引擎路由加载失败: {}".format(e))
# ============================================================
# FT-Transformer 量化训练系统路由
# ============================================================
try:
_ft_routes_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'output', 'ft_train_routes.py')
import importlib.util as _ilu_ft
_ft_spec = _ilu_ft.spec_from_file_location('ft_train_routes', _ft_routes_path)
_ft_mod = _ilu_ft.module_from_spec(_ft_spec)
_ft_spec.loader.exec_module(_ft_mod)
_ft_mod.register_routes(app)
print("[INFO] FT-Transformer训练路由已加载")
except Exception as e:
print("[WARN] FT-Transformer训练路由加载失败: {}".format(e))
# ============================================================
# F-v3 独立预测端点 (数据驱动6因子方向修正, 准确率55.5%)
# ============================================================
@app.route('/api/fv3-analysis', methods=['POST'])
def fv3_analysis():
"""方案F-v5: MACD(5,10,3)+高置信模式(BUY>70/SELL<28)+6因子方向修正
回测准确率55.5% (47信号), BUY=56.0%, SELL=86.4%
6因子: MACD(+1), KDJ(+1), VWAP(+1), MA(-1翻转), VP(-1翻转), RSI(-1翻转)
等权各1/6, 阈值BUY>62 SELL<38
"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
if not V53_SCORER_AVAILABLE:
return jsonify({'error': 'F-v5模块未加载'}), 503
df, err = _download_stock_data(stock_code)
if err:
return jsonify({'error': err}), 400
if df is None:
return jsonify({'error': '无数据'}), 400
if 'timestamps' in df.columns:
df = df.set_index('timestamps')
fv3_result = _v53_mod.run_fv3_analysis(stock_code, df)
return jsonify({
'success': True,
'stock_code': stock_code,
'data': {
'fv5_score': fv3_result.get('score', 50),
'fv5_signal': fv3_result.get('signal', 'HOLD'),
'fv5_confidence': fv3_result.get('confidence', 50),
'fv5_tier': fv3_result.get('tier', 'HOLD'),
'fv5_label': fv3_result.get('label', ''),
'plan_fv5': {
'signal': fv3_result.get('signal', 'HOLD'),
'trend_direction': {'BUY': '看涨', 'SELL': '看跌', 'HOLD': '震荡'}.get(fv3_result.get('signal', 'HOLD'), '震荡'),
'confidence': fv3_result.get('confidence', 50),
'score': fv3_result.get('score', 50),
'tier': fv3_result.get('tier', 'HOLD'),
'source': 'plan_f_v3',
'note': '方案F-v5: MACD(5,10,3)+高置信模式+方向修正(MACD+KDJ+VWAP正向,MA+VP+RSI反向) 准确率55.5%',
'label': fv3_result.get('label', ''),
'details': fv3_result.get('plan_f', {}),
},
'ft_prediction': {
'signal': fv3_result.get('signal', 'HOLD'),
'trend_direction': {'BUY': '看涨', 'SELL': '看跌', 'HOLD': '震荡'}.get(fv3_result.get('signal', 'HOLD'), '震荡'),
'confidence': fv3_result.get('confidence', 50),
'score': fv3_result.get('score', 50),
'tier': fv3_result.get('tier', 'HOLD'),
'source': 'plan_f_v3',
'note': '方案F-v5: MACD(5,10,3)+高置信模式+方向修正(MACD+KDJ+VWAP正向,MA+VP+RSI反向) 准确率55.5%',
},
}
})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': f'F-v5分析失败: {str(e)}'}), 500
@app.route('/api/ml-analysis', methods=['POST'])
def ml_analysis():
"""蒸鱼ML预测引擎: GBDT+逻辑回归双模型日线预测
GBDT准确率54.0% (16股600日数据重新训练)
"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
if not ML_ENGINE_AVAILABLE:
return jsonify({'error': 'ML引擎未加载'}), 503
ml_result = _ml_mod.predict(stock_code)
if ml_result.get('error'):
return jsonify({'error': ml_result['error']}), 400
signal = ml_result['signal']
confidence = ml_result['confidence']
# 信号等级
if signal == 'BUY' and ml_result['high_confidence']:
tier = 'STRONG_BUY'
elif signal == 'SELL' and ml_result['high_confidence']:
tier = 'STRONG_SELL'
elif signal == 'BUY':
tier = 'BUY'
elif signal == 'SELL':
tier = 'SELL'
else:
tier = 'HOLD'
return jsonify({
'success': True,
'stock_code': stock_code,
'data': {
'ml_signal': signal,
'ml_confidence': confidence,
'ml_tier': tier,
'ml_high_confidence': ml_result['high_confidence'],
'ml_gbdt_prob': ml_result['gbdt_prob_buy'],
'ml_lr_prob': ml_result['lr_prob_buy'],
'ml_note': '蒸鱼ML引擎: GBDT 54.0% (16股600日重新训练)',
'ml_features': ml_result.get('features', {}),
}
})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': f'ML分析失败: {str(e)}'}), 500
@app.route('/api/ml-daily', methods=['POST'])
def ml_daily():
"""获取ML训练用的日线数据(供前端展示)"""
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
d = _ml_mod.download_daily(stock_code)
if d is None:
return jsonify({'error': f'无法获取{stock_code}的日线数据'}), 400
c = d['close']; h = d['high']; l = d['low']; v = d['volume']
days = []
for i in range(len(c)):
days.append({
'idx': i,
'close': round(float(c[i]), 2),
'high': round(float(h[i]), 2),
'low': round(float(l[i]), 2),
'volume': round(float(v[i]), 0),
})
return jsonify({'success': True, 'stock_code': stock_code,
'count': len(days), 'days': days[-120:]})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/ml-train', methods=['POST'])
def ml_train():
"""ML模型训练: GBDT+LR重新训练(16只股票, 约1-2分钟)"""
import threading
global _ml_training, _ml_train_result
if not ML_ENGINE_AVAILABLE:
return jsonify({'error': 'ML引擎未加载'}), 503
if getattr(ml_train, '_training', False):
return jsonify({'status': 'training', 'message': '模型训练中，请稍候...'})
def _do_train():
try:
ml_train._training = True
result = _ml_mod.train_and_save_models(force=True)
ml_train._result = {'status': 'completed', 'success': result,
'message': '训练完成!' if result else '训练失败: 数据不足'}
except Exception as e:
ml_train._result = {'status': 'error', 'message': str(e)}
finally:
ml_train._training = False
ml_train._training = True
ml_train._result = None
t = threading.Thread(target=_do_train, daemon=True)
t.start()
return jsonify({'status': 'training', 'message': '训练已启动，16只股票约1-2分钟...'})
@app.route('/api/ml-train/status', methods=['GET'])
def ml_train_status():
"""查询ML训练状态"""
training = getattr(ml_train, '_training', False)
result = getattr(ml_train, '_result', None)
return jsonify({
'status': 'training' if training else (result.get('status') if result else 'idle'),
'message': result.get('message') if result else '就绪',
'success': result.get('success') if result else None,
})
# ================================================================
# 多维因子评分 & 因子市场状态 (QuantGPT适配)
# ================================================================
@app.route('/api/factor-score', methods=['POST'])
def api_factor_score():
"""多维因子评分：6维度综合评分（IC/IR/稳定性/反过拟合/回测/Cloud对齐）"""
data = request.get_json(force=True) or {}
stock_code = data.get('code', '') or data.get('stock_code', '600519')
count = int(data.get('count', 500))
try:
# 获取K线数据
klines = _fetch_klines(stock_code, count)
if not klines or len(klines) < 60:
return jsonify({'success': False, 'error': f'{stock_code}数据不足'})
# 运行多因子评估
from factor_score import compute_factor_score, FACTOR_EVAL_PRESETS
results = []
for key, preset in FACTOR_EVAL_PRESETS.items():
try:
from factor_engine import parse_expression
# 使用腾讯API格式的列名（腾讯qfqday可能6列或7列）
import pandas as pd
cleaned = []
for k in klines:
if isinstance(k, list) and len(k) >= 6:
cleaned.append(k[:6]) # 截断到6列: date,open,close,high,low,volume
elif isinstance(k, str):
parts = k.split(',')
if len(parts) >= 6:
cleaned.append(parts[:6])
cols = ['date', 'open', 'close', 'high', 'low', 'volume']
df = pd.DataFrame(cleaned, columns=cols)
for c in ['open', 'close', 'high', 'low', 'volume']:
df[c] = pd.to_numeric(df[c], errors='coerce')
factor_vals = parse_expression(preset['expression'], df)
if isinstance(factor_vals, pd.Series):
factor_vals = factor_vals.values
# 简化的IC计算
rets = df['close'].pct_change().shift(-1).dropna().values
fv = factor_vals[1:len(rets)+1]
valid = np.isfinite(fv) & np.isfinite(rets)
if valid.sum() < 20:
ic_mean = 0
ic_ir = 0
ic_win_rate = 0
else:
daily_ics = []
for i in range(0, len(fv)-20, 5):
window_fv = fv[i:i+20]
window_ret = rets[i:i+20]
v = np.isfinite(window_fv) & np.isfinite(window_ret)
if v.sum() >= 10:
c = np.corrcoef(window_fv[v], window_ret[v])[0,1]
if np.isfinite(c):
daily_ics.append(c)
if daily_ics:
ic_arr = np.array(daily_ics)
ic_mean = float(np.mean(ic_arr))
ic_std = float(np.std(ic_arr))
ic_ir = ic_mean / ic_std if ic_std > 0 else 0
ic_win_rate = float(np.mean(np.array(daily_ics) > 0))
# IC合理性检查: 单股时序IC超过0.15为异常（暴涨暴跌噪声），截断
max_ic = 0.15
if abs(ic_mean) > max_ic:
ic_mean = max_ic if ic_mean > 0 else -max_ic
if abs(ic_ir) > 5.0:
ic_ir = 5.0 if ic_ir > 0 else -5.0
else:
ic_mean = ic_ir = ic_win_rate = 0
# 计算评分
bt_mock = {
'success': True,
'summary': {
'ic_mean': ic_mean, 'ic_ir': ic_ir,
'ic_win_rate': ic_win_rate,
},
'performance': {
'long_short': {'sharpe': abs(ic_ir) * 0.8},
'monotonicity': 0.3,
'group_1': {'sharpe': ic_ir},
'group_5': {'sharpe': -ic_ir},
'spread': ic_mean * 2,
},
}
score = compute_factor_score(bt_mock)
results.append({
'name': preset['name'],
'category': preset['category'],
'expression': preset['expression'],
'score': score,
'ic_mean': round(ic_mean, 6),
'ic_ir': round(ic_ir, 4),
'ic_win_rate': round(ic_win_rate, 4),
})
except Exception as e:
results.append({
'name': preset['name'], 'category': preset['category'],
'error': str(e), 'score': {'overall': 0, 'grade': 'N/A', 'dimensions': {}},
})
# 按评分排序
results.sort(key=lambda x: x['score'].get('overall', 0), reverse=True)
return jsonify({'success': True, 'stock_code': stock_code,
'n_factors': len(results), 'results': results})
except Exception as e:
import traceback
return jsonify({'success': False, 'error': str(e), 'traceback': traceback.format_exc()})
@app.route('/api/factor-regime', methods=['POST'])
def api_factor_regime():
"""从因子信号推导市场状态：regime + style + risk_level + headline"""
data = request.get_json(force=True) or {}
stock_code = data.get('code', '600519')
count = int(data.get('count', 500))
try:
klines = _fetch_klines(stock_code, count)
if not klines or len(klines) < 30:
return jsonify({'success': False, 'error': '数据不足'})
import pandas as pd
from factor_regime import detect_factor_regime, get_regime_advice
cols = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount']
df = pd.DataFrame(klines, columns=cols)
for c in ['open', 'close', 'high', 'low', 'volume', 'amount']:
df[c] = pd.to_numeric(df[c], errors='coerce')
regime = detect_factor_regime(
close=df['close'].values,
high=df['high'].values,
low=df['low'].values,
volume=df['volume'].values,
)
advice = get_regime_advice(regime)
return jsonify({'success': True, 'stock_code': stock_code,
'regime': regime, 'advice': advice})
except Exception as e:
import traceback
return jsonify({'success': False, 'error': str(e), 'traceback': traceback.format_exc()})
@app.route('/api/market-regime-composite', methods=['POST'])
def api_market_regime_composite():
"""综合市场状态：融合ADX原始regime + 因子推导regime"""
data = request.get_json(force=True) or {}
stock_code = data.get('code', '600519')
count = int(data.get('count', 500))
try:
klines = _fetch_klines(stock_code, count)
if not klines or len(klines) < 60:
return jsonify({'success': False, 'error': '数据不足'})
import pandas as pd
cols = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount']
df = pd.DataFrame(klines, columns=cols)
for c in ['open', 'close', 'high', 'low', 'volume', 'amount']:
df[c] = pd.to_numeric(df[c], errors='coerce')
# ADX Regime（现有模块）
from market_regime import detect_market_regime as _detect_adx_regime
adx_regime = _detect_adx_regime(
df['close'].values, df['high'].values,
df['low'].values, df['volume'].values
)
# 因子Regime（新增模块）
from factor_regime import detect_factor_regime, get_regime_advice as _get_fregime_advice
factor_regime = detect_factor_regime(
close=df['close'].values, high=df['high'].values,
low=df['low'].values, volume=df['volume'].values
)
# 融合
composite = {
'success': True,
'stock_code': stock_code,
'adx_regime': {
'regime': adx_regime.get('regime', 'RANGE'),
'label': adx_regime.get('regime_label', '未知'),
'adx': round(adx_regime.get('adx', 0), 2),
'volatility': adx_regime.get('volatility', 'NORMAL'),
'atr_pct': round(adx_regime.get('atr_pct', 0), 4),
},
'factor_regime': {
'regime': factor_regime['regime'],
'style': factor_regime['style'],
'risk_level': factor_regime['risk_level'],
'headline': factor_regime['headline'],
'scores': factor_regime['scores'],
},
'consensus': _build_consensus(adx_regime, factor_regime),
'advice': _get_fregime_advice(factor_regime),
}
return jsonify(composite)
except Exception as e:
import traceback
return jsonify({'success': False, 'error': str(e), 'traceback': traceback.format_exc()})
def _build_consensus(adx_regime, factor_regime):
"""融合两种regime得到共识"""
adx_r = adx_regime.get('regime', 'RANGE')
factor_r = factor_regime.get('regime', '震荡市')
# 映射到统一分类
def map_regime(r):
if r in ('TREND_UP', 'TREND_DOWN'):
return '趋势'
elif r in ('高波动',):
return '高波动'
return '震荡'
mapped_adx = map_regime(adx_r)
mapped_factor = map_regime(factor_r)
if mapped_adx == mapped_factor:
consensus = mapped_adx
confidence = '高'
elif mapped_adx == '趋势' and mapped_factor == '高波动':
consensus = '趋势高波动'
confidence = '中'
elif mapped_factor == '高波动':
consensus = '高波动'
confidence = '中'
else:
consensus = '震荡偏趋势'
confidence = '低'
return {
'regime': consensus,
'confidence': confidence,
'adx_vote': mapped_adx,
'factor_vote': mapped_factor,
}
def _fetch_klines(code, count=500):
"""获取K线数据（腾讯API）"""
import urllib.request
tc = ('sh' + code.zfill(6)) if code.startswith('6') else ('sz' + code.zfill(6))
url = f"http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={tc},day,,,{count},qfq"
try:
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
resp = urllib.request.urlopen(req, timeout=10)
data = json.loads(resp.read().decode('utf-8'))
return data.get('data', {}).get(tc, {}).get('qfqday', [])
except Exception:
return []
# ============================================================
# 🏆 蒸鱼终极评分 API
# ============================================================
import importlib.util as _ilu_ultimate
_ultimate_available = False
try:
_ult_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'ultimate_scorer.py')
_ult_spec = _ilu_ultimate.spec_from_file_location('ultimate_scorer', _ult_path)
_ult_mod = _ilu_ultimate.module_from_spec(_ult_spec)
sys.modules['ultimate_scorer'] = _ult_mod
_ult_spec.loader.exec_module(_ult_mod)
from ultimate_scorer import get_ultimate_scorer
_ultimate_available = True
print("[OK] 终极评分引擎已加载")
except Exception as e:
print(f"[Warning] 终极评分引擎加载失败: {e}")
# ====== 融合引擎 v6.1 (含目标价格计算) ======
try:
import fusion_v2 as _fusion_v2
_FUSION_V2_AVAILABLE = True
print("[OK] 融合引擎 v6.1 已加载 (含目标价格计算)")
except Exception as e:
_FUSION_V2_AVAILABLE = False
print(f"[Warning] 融合引擎 v6.1 加载失败: {e}")
@app.route('/api/ultimate-score', methods=['POST'])
def ultimate_score():
"""终极评分: 汇聚所有分析引擎, 基于回测准确率加权, 给出终极评判"""
if not _ultimate_available:
return jsonify({'error': '终极评分引擎未加载'}), 500
try:
data = request.get_json()
stock_code = data.get('stock_code', '').strip()
if not stock_code:
return jsonify({'error': '请输入股票代码'}), 400
scorer = get_ultimate_scorer()
engine_results = {}
# 1. 下载一次数据供多个引擎共用
df, err = _download_stock_data(stock_code)
if err or df is None:
return jsonify({'error': f'获取数据失败: {err}'}), 400
# 2. 收集各引擎结果
# --- 威科夫量价 ---
try:
if WYCKOFF_VPA_AVAILABLE:
df_wv = df.tail(250).reset_index(drop=True)
wv_result = detect_wyckoff_vpa(df_wv)
engine_results['wyckoff_vpa'] = wv_result
except Exception as e:
engine_results['wyckoff_vpa_error'] = str(e)[:100]
# --- 经典K线形态 ---
try:
if CLASSIC_PATTERN_AVAILABLE:
cp_result = detect_classic_patterns(df)
engine_results['classic_patterns'] = cp_result
except Exception as e:
engine_results['classic_patterns_error'] = str(e)[:100]
# --- 波浪理论 ---
try:
if WAVE_ANALYSIS_AVAILABLE:
wa_result = analyze_wave_patterns(df)
engine_results['wave'] = wa_result
except Exception as e:
engine_results['wave_error'] = str(e)[:100]
# --- 多维度联动 ---
try:
if ENHANCED_AVAILABLE and multidim_analyzer is not None:
md_result = multidim_analyzer.analyze(df.tail(250).reset_index(drop=True))
engine_results['multi_dimension'] = md_result
except Exception as e:
engine_results['multi_dimension_error'] = str(e)[:100]
# --- V4 五维综合 ---
try:
v4_result = _v4_mod.run_v4_composite_analysis(stock_code, df)
engine_results['v4_composite'] = v4_result
except Exception as e:
engine_results['v4_error'] = str(e)[:100]
# --- F-v5 ---
try:
if V53_SCORER_AVAILABLE:
fv5_result = _v53_mod.run_fv3_analysis(stock_code, df)
engine_results['fv5'] = fv5_result
except Exception as e:
engine_results['fv5_error'] = str(e)[:100]
# --- 方案E+ (从joint-prediction提取) ---
try:
import json as _json
# 简化版: 直接调用joint_prediction中的逻辑
if V5_SCORER_AVAILABLE:
plan_e_result = _compute_plan_e(stock_code, df)
if plan_e_result:
engine_results['plan_e'] = plan_e_result
except Exception as e:
engine_results['plan_e_error'] = str(e)[:100]
# --- ML引擎 ---
try:
if ML_ENGINE_AVAILABLE:
ml_result = _ml_engine_analyze(stock_code, df)
engine_results['ml_engine'] = ml_result
except Exception as e:
engine_results['ml_error'] = str(e)[:100]
# --- 踏浪之道 ---
try:
if TALANG_AVAILABLE:
from talang_integration import talang_score_analysis
tl_result = talang_score_analysis(df)
# 统一字段名供终极评分提取: score/max_score/signal
if tl_result and isinstance(tl_result, dict):
tl_result['score'] = tl_result.get('latest_score', tl_result.get('score', 0))
tl_result['max_score'] = tl_result.get('latest_scores', {})
# 如果latest_scores是dict, max_score=15(规则总数)
if isinstance(tl_result['max_score'], dict):
tl_result['max_score'] = 15
tl_result['signal'] = tl_result.get('latest_signal', tl_result.get('signal', '中性'))
# buy/sell markers计数 (供v2映射使用)
tl_result['buy_markers_count'] = len(tl_result.get('buy_markers', []))
tl_result['sell_markers_count'] = len(tl_result.get('sell_markers', []))
engine_results['talang'] = tl_result
except Exception as e:
engine_results['talang_error'] = str(e)[:100]
# --- 因子评分 (构造回测结果字典供compute_factor_score使用) ---
try:
from factor_score import compute_factor_score
# compute_factor_score需要backtest_result字典, 非DataFrame
# 从df计算IC等指标构造兼容的回测结果
import talib as _talib
df_fa = df.tail(250).reset_index(drop=True)
close_arr = df_fa['close'].values.astype(float)
ret_1d = np.diff(close_arr) / close_arr[:-1]
# 计算IC (因子=动量因子: 过去5日收益率)
mom5 = np.zeros(len(close_arr))
for _i in range(5, len(close_arr)):
mom5[_i] = (close_arr[_i] - close_arr[_i-5]) / close_arr[_i-5]
ic_raw = np.corrcoef(mom5[6:], ret_1d[5:])[0,1] if len(ret_1d[5:]) > 10 else 0
ic_series = np.array([ic_raw * (0.8 + 0.2 * np.random.randn()) for _ in range(12)])
# 构造backtest_result字典
factor_backtest = {
'success': True,
'summary': {
'ic_mean': abs(float(ic_raw)),
'ic_ir': abs(float(ic_raw)) / (np.std(ic_series) + 0.001),
'ic_win_rate': float(np.mean(np.sign(ic_series) == np.sign(ic_raw))),
},
'performance': {
'group_1': {'sharpe': float(ic_raw * 3)},
'group_5': {'sharpe': float(-ic_raw * 2)},
'long_short': {'sharpe': abs(float(ic_raw)) * 2.5},
'monotonicity': 0.7 if abs(ic_raw) > 0.02 else 0.3,
'spread': abs(float(ic_raw)) * 0.8,
},
}
factor_result = compute_factor_score(factor_backtest)
engine_results['factor'] = factor_result
except Exception as e:
import traceback as _tb
_tb.print_exc()
engine_results['factor_error'] = str(e)[:100]
# 3. 终极评分
result = scorer.analyze(engine_results)
# 4. 融合引擎 v6.1 (含目标价格计算)
fusion_result = None
if _FUSION_V2_AVAILABLE:
try:
fusion_result = _fusion_v2.fuse(result, df)
except Exception as _fe:
print(f"[Fusion] 融合计算失败: {_fe}")
return jsonify({
'success': True,
'stock_code': stock_code,
'ultimate': result,
'fusion': fusion_result,
})
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
def _compute_plan_e(stock_code, df):
"""计算方案E+信号 (从joint-prediction中抽取核心逻辑)"""
try:
if not V5_SCORER_AVAILABLE:
return None
df_v5 = df.copy()
if 'timestamps' in df_v5.columns:
df_v5 = df_v5.set_index('timestamps')
v5_result = _v5_mod.run_v5_optimized_analysis(stock_code, df_v5)
# 市场状态
regime_result = None
adaptive_params = {}
try:
from market_regime import detect_market_regime, get_adaptive_params as _gap
regime_result = detect_market_regime(
df['close'].values.astype(float),
df['high'].values.astype(float),
df['low'].values.astype(float),
df['volume'].values.astype(float)
)
adaptive_params = _gap(regime_result)
except:
pass
v5_sc = v5_result.get('score', 50)
v5_sig = v5_result.get('signal', 'HOLD')
a_boost = adaptive_params.get('signal_boost', 1.0)
a_v4w = adaptive_params.get('v4_vote_weight', 1.5)
return {
'signal': v5_sig,
'confidence': 50,
'score': v5_sc,
}
except Exception as e:
return None
def _ml_engine_analyze(stock_code, df):
"""ML引擎分析 - 直接调用predict(stock_code)"""
try:
from ml_engine import predict
result = predict(stock_code)
if isinstance(result, dict):
return result
return {}
except Exception as e:
return {'error': str(e)}
@app.route('/api/fusion-backtest-report')
def fusion_backtest_report():
"""融合回测验证报告 - 读取已保存的QuantMind V6.1报告"""
import json, os
base = os.path.dirname(os.path.abspath(__file__))
report_path = os.path.join(base, 'output', 'fusion_backtest_report.json')
if not os.path.exists(report_path):
return jsonify({'success': False, 'error': '回测报告未生成，请先运行融合回测验证'})
try:
with open(report_path, 'r', encoding='utf-8') as f:
data = json.load(f)
return jsonify({'success': True, 'report': data})
except Exception as e:
return jsonify({'success': False, 'error': str(e)})
if __name__ == '__main__':
port = int(os.getenv('PORT', '7860'))
debug = os.getenv('FLASK_DEBUG', 'false').lower() == 'true'
app.run(debug=debug, host='0.0.0.0', port=port)