app.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
升级版 app.py — 高精度数值 / 高性能统计 + 精细化 LLM 策略输出
功能亮点：
- Crank–Nicolson PDE（Black–Scholes）
- Monte Carlo：Antithetic + Control variates（使用 BS 解析作为控制变量）
- GARCH(1,1) 使用 arch （若可用）或 MLE minimize 回退
- Johansen 协整检验（statsmodels 若可用）
- 组合优化使用 cvxpy（若可用）或 SciPy 回退
- LLM 生成结构化 JSON 策略（策略说明、信号、伪代码、回测/风险提示）
- 保持之前的几何/Whitney/Noise/Gradient 模块兼容
"""

import os
import json
import warnings
warnings.filterwarnings("ignore")

from pathlib import Path
from datetime import datetime
from typing import Any, Dict, Optional, Tuple, List

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

# torch used for embedding / potential LSTM
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.optim import Adam

# statsmodels optional
try:
    import statsmodels.api as sm
    from statsmodels.tsa.vector_ar.vecm import coint_johansen
    from statsmodels.tsa.api import VAR
    STATS_MODELS_AVAILABLE = True
except Exception:
    STATS_MODELS_AVAILABLE = False

# arch package (GARCH) optional
try:
    from arch import arch_model
    ARCH_AVAILABLE = True
except Exception:
    ARCH_AVAILABLE = False

# cvxpy for portfolio optimization optional
try:
    import cvxpy as cp
    CVXPY_AVAILABLE = True
except Exception:
    CVXPY_AVAILABLE = False

# scipy fallback utilities
from scipy import stats
from scipy.optimize import minimize
from scipy.linalg import toeplitz

# HTTP for LLM
import requests

# Gradio UI
import gradio as gr

# Logging
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("quant_upgraded")

# base dir
BASE_DIR = Path("/tmp/quant_upgraded")
BASE_DIR.mkdir(parents=True, exist_ok=True)

# ---------------------
# Configuration
# ---------------------
class Config:
    def __init__(self):
        self.device = 'cpu'
        if torch.cuda.is_available():
            self.device = 'cuda'
        self.hf_token = os.getenv("HF_API_TOKEN", "")
        self.hf_default_model = "Qwen/Qwen2.5-1.5B-Instruct"
        self.mc_default_paths = 20000
        self.cv_solver = "cvxpy" if CVXPY_AVAILABLE else "scipy"
        self.statsmodels = STATS_MODELS_AVAILABLE
        self.arch = ARCH_AVAILABLE

config = Config()

# ---------------------
# Geometry / existing modules (compact)
# ---------------------
class FiberBundleTheory:
    def __init__(self, fiber_dim=16):
        self.fiber_dim = fiber_dim
        self.whitney_factor = 2 * fiber_dim

    def project_to_base(self, x: np.ndarray) -> np.ndarray:
        x = np.asarray(x).ravel()
        if len(x) < self.fiber_dim:
            x = np.pad(x, (0, self.fiber_dim - len(x)))
        half = self.fiber_dim // 2
        vix2 = float(np.sum(x[:half]**2) / (half + 1e-12))
        rv = float(np.std(x[half:]))
        return np.array([vix2, rv])

    def compute_vrp(self, base_point: np.ndarray) -> float:
        vix2, rv = base_point
        return vix2 - rv

class NoiseExplorer:
    def regress_vix2_vs_rv(self, vix2: np.ndarray, rv: np.ndarray):
        X = np.vstack([rv, np.ones_like(rv)]).T
        coef, *_ = np.linalg.lstsq(X, vix2, rcond=None)
        a, b = float(coef[0]), float(coef[1])
        preds = a * rv + b
        resid = vix2 - preds
        return {'a': a, 'b': b, 'preds': preds, 'resid': resid}

    def resid_stats(self, resid: np.ndarray):
        resid = np.asarray(resid)
        mean = float(np.mean(resid))
        var = float(np.var(resid))
        ac1 = float(np.corrcoef(resid[:-1], resid[1:])[0,1]) if len(resid) > 2 else 0.0
        fft = np.fft.rfft(resid - mean)
        freqs = np.fft.rfftfreq(len(resid))
        power = np.abs(fft)**2
        dominant_freq = float(freqs[np.argmax(power[1:])+1]) if len(power) > 1 else 0.0
        return {'mean': mean, 'var': var, 'ac1': ac1, 'dominant_freq': dominant_freq}

    def explore(self, df: pd.DataFrame, vix2_col: Optional[str]=None, rv_col: Optional[str]=None):
        numcols = df.select_dtypes(include=[np.number]).columns.tolist()
        if not numcols:
            return None
        if vix2_col is None or rv_col is None:
            vix2_col = numcols[0]
            rv_col = numcols[1] if len(numcols) > 1 else numcols[0]
        vix2 = df[vix2_col].fillna(method='ffill').values
        rv = df[rv_col].fillna(method='ffill').values
        reg = self.regress_vix2_vs_rv(vix2, rv)
        st = self.resid_stats(reg['resid'])
        vrp = vix2 - reg['preds']
        return {
            'vix2_col': vix2_col,
            'rv_col': rv_col,
            'reg': {'a': reg['a'], 'b': reg['b']},
            'resid_stats': st,
            'vrp_mean': float(np.mean(vrp)),
            'vrp_std': float(np.std(vrp)),
            'vrp_series': vrp.tolist(),
            'residuals': reg['resid'].tolist()
        }

# ---------------------
# Quant modules (upgraded)
# ---------------------

class StochasticModels:
    """High-precision stochastic processes and pricing helpers."""

    @staticmethod
    def bs_price(S: float, K: float, r: float, q: float, sigma: float, T: float, option_type: str='call') -> float:
        """Black-Scholes closed-form price (with dividend yield q)."""
        S, K, r, q, sigma, T = map(float, (S, K, r, q, sigma, T))
        if T <= 0 or sigma <= 0:
            return float(max(S - K, 0.0) if option_type == 'call' else max(K - S, 0.0))
        d1 = (np.log(S / K) + (r - q + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
        d2 = d1 - sigma * np.sqrt(T)
        if option_type == 'call':
            price = S * np.exp(-q * T) * stats.norm.cdf(d1) - K * np.exp(-r * T) * stats.norm.cdf(d2)
        else:
            price = K * np.exp(-r * T) * stats.norm.cdf(-d2) - S * np.exp(-q * T) * stats.norm.cdf(-d1)
        return float(price)

    @staticmethod
    def heston_simulate(S0: float, v0: float, r: float, kappa: float, theta: float, xi: float, rho: float, T: float,
                       n_steps: int=252, n_paths: int=2000, seed: Optional[int]=None):
        """
        Euler-Maruyama with full-reflection for variance (CIR-like) — more stable by forcing v>=0.
        Keep path count moderate unless GPU simulation used externally.
        """
        if seed is not None:
            np.random.seed(seed)
        dt = T / n_steps
        S = np.zeros((n_paths, n_steps+1))
        v = np.zeros((n_paths, n_steps+1))
        S[:,0] = S0
        v[:,0] = v0
        for t in range(n_steps):
            z1 = np.random.randn(n_paths)
            z2 = np.random.randn(n_paths)
            w1 = z1
            w2 = rho * z1 + np.sqrt(max(0.0, 1 - rho**2)) * z2
            v_prev = np.maximum(v[:,t], 0.0)
            # full truncation Euler
            dv = kappa * (theta - v_prev) * dt + xi * np.sqrt(v_prev * dt) * w2
            v_new = np.maximum(v_prev + dv, 1e-8)
            dS = r * S[:,t] * dt + np.sqrt(v_prev * dt) * S[:,t] * w1
            S[:,t+1] = S[:,t] + dS
            v[:,t+1] = v_new
        return S, v

    @staticmethod
    def merton_jump_diffusion(S0: float, mu: float, sigma: float, lamb: float, mu_j: float, sigma_j: float,
                              T: float, n_steps: int=252, n_paths: int=2000, seed: Optional[int]=None):
        """Improved Merton simulator with vectorized operations."""
        if seed is not None:
            np.random.seed(seed)
        dt = T / n_steps
        S = np.full((n_paths, n_steps+1), S0, dtype=float)
        for t in range(n_steps):
            z = np.random.randn(n_paths)
            pois = np.random.poisson(lamb * dt, size=n_paths)
            jumps = np.exp(mu_j + sigma_j * np.random.randn(n_paths)) - 1.0
            S[:, t+1] = S[:, t] * (1 + mu*dt + sigma*np.sqrt(dt)*z) + S[:, t] * (jumps * pois)
            S[:, t+1] = np.maximum(S[:, t+1], 1e-8)
        return S

class NumericalMethods:
    """Crank-Nicolson PDE + Monte Carlo with variance reduction."""

    @staticmethod
    def bs_crank_nicolson(S0: float, K: float, r: float, q: float, sigma: float, T: float,
                          Smax_mult: float=3.0, M: int=400, N: int=400, option_type: str='call') -> float:
        """
        Crank-Nicolson solver for Black-Scholes PDE. More stable with sufficient grid resolution.
        M: number of asset steps, N: time steps.
        """
        Smax = S0 * Smax_mult
        dS = Smax / M
        dt = T / N
        grid = np.zeros((M+1, N+1))
        Svals = np.linspace(0, Smax, M+1)
        # terminal condition
        if option_type == 'call':
            grid[:, -1] = np.maximum(Svals - K, 0)
        else:
            grid[:, -1] = np.maximum(K - Svals, 0)
        # boundary conditions
        grid[0, :] = 0.0 if option_type == 'call' else K * np.exp(-r * (T - np.linspace(0, T, N+1)))
        grid[-1, :] = (Smax - K * np.exp(-r * (T - np.linspace(0, T, N+1)))) if option_type == 'call' else 0.0
        # prepare tridiagonal coefficients
        j = np.arange(1, M)
        a = 0.25 * dt * (sigma**2 * j**2 - (r - q) * j)
        b = -0.5 * dt * (sigma**2 * j**2 + r)
        c = 0.25 * dt * (sigma**2 * j**2 + (r - q) * j)
        # construct A and B matrices (tridiagonal)
        A = np.zeros((M-1, M-1))
        B = np.zeros((M-1, M-1))
        for idx in range(M-1):
            if idx > 0:
                A[idx, idx-1] = -a[idx+1]
                B[idx, idx-1] = a[idx+1]
            A[idx, idx] = 1 - b[idx+1]
            B[idx, idx] = 1 + b[idx+1]
            if idx < M-2:
                A[idx, idx+1] = -c[idx+1]
                B[idx, idx+1] = c[idx+1]
        # backward time stepping
        from numpy.linalg import solve
        for n in reversed(range(N)):
            rhs = B.dot(grid[1:M, n+1])
            # add boundary contributions
            rhs[0] += a[1] * (grid[0, n] + grid[0, n+1])
            rhs[-1] += c[M-1] * (grid[M, n] + grid[M, n+1])
            grid[1:M, n] = solve(A, rhs)
        # interpolate at S0
        i = int(S0 / dS)
        if i >= M:
            return float(grid[-1, 0])
        w = (S0 - i * dS) / dS
        price = (1-w) * grid[i, 0] + w * grid[i+1, 0]
        return float(price)

    @staticmethod
    def mc_price_bs_cv(S0: float, K: float, r: float, q: float, sigma: float, T: float,
                       option_type: str='call', n_paths: int=20000, antithetic: bool=True, seed: Optional[int]=None):
        """
        Monte Carlo with antithetic variates and control variate (BS analytic).
        Control variate: use discount payoff under geometric Brownian motion analytic expectation = BS price with same params.
        """
        if seed is not None:
            np.random.seed(seed)
        n = n_paths
        half = n // 2 if antithetic else n
        Z = np.random.randn(half)
        if antithetic:
            Z = np.concatenate([Z, -Z])
        ST = S0 * np.exp((r - q - 0.5*sigma**2) * T + sigma * np.sqrt(T) * Z)
        if option_type == 'call':
            payoff = np.maximum(ST - K, 0)
        else:
            payoff = np.maximum(K - ST, 0)
        # control variate: use discounted ST (or log ST) expectation known
        # use analytic BS price as control target
        bs_analytic = StochasticModels.bs_price(S0, K, r, q, sigma, T, option_type=option_type)
        # choose control variable as discounted payoff under geometric mean? simple: use ST
        control = ST  # expectation of ST under risk-neutral = S0 * exp((r-q)T)
        control_mean = S0 * np.exp((r - q) * T)
        # compute covariance and adjust
        cov_pc = np.cov(payoff, control, ddof=1)[0,1]
        var_c = np.var(control, ddof=1)
        if var_c > 0:
            beta = cov_pc / var_c
        else:
            beta = 0.0
        adj_payoff = payoff - beta * (control - control_mean)
        price = np.exp(-r * T) * np.mean(adj_payoff)
        # bias correction via analytic price difference if helpful
        return float(price)

class Econometrics:
    """GARCH via arch package (preferred) or MLE fallback; Johansen using statsmodels if available"""

    @staticmethod
    def garch_11_fit(returns: np.ndarray):
        r = np.asarray(returns).astype(float)
        r = r - np.mean(r)
        if config.arch:
            try:
                am = arch_model(r * 100.0, vol='Garch', p=1, q=1, dist='normal')  # scale to percent to help arch convergence
                res = am.fit(disp='off')
                params = res.params.to_dict()
                cond_var = res.conditional_volatility / 100.0
                return {'method': 'arch', 'params': params, 'cond_var': cond_var.tolist()}
            except Exception as e:
                logger.warning(f"arch fit failed: {e}; falling back to MLE.")
        # MLE fallback
        T = len(r)
        def neglog(params):
            omega, alpha, beta = params
            if omega <= 0 or alpha < 0 or beta < 0 or alpha + beta >= 0.9999:
                return 1e12
            h = np.zeros(T)
            h[0] = np.var(r)
            ll = 0.0
            for t in range(1, T):
                h[t] = omega + alpha * r[t-1]**2 + beta * h[t-1]
            ll = 0.5 * (np.log(2*np.pi) + np.log(h) + (r**2)/h)
            return np.sum(ll)
        init = np.array([1e-6, 0.05, 0.9])
        bnds = [(1e-12, None), (0, 0.9999), (0, 0.9999)]
        res = minimize(neglog, x0=init, bounds=bnds)
        if not res.success:
            logger.warning("GARCH MLE did not converge; returning fallback params")
            omega, alpha, beta = init
        else:
            omega, alpha, beta = res.x
        # compute h
        h = np.zeros(T)
        h[0] = np.var(r)
        for t in range(1, T):
            h[t] = omega + alpha * r[t-1]**2 + beta * h[t-1]
        return {'method': 'mle', 'params': {'omega': float(omega), 'alpha': float(alpha), 'beta': float(beta)}, 'cond_var': h.tolist()}

    @staticmethod
    def johansen_test(data: np.ndarray, det_order: int=0, k_ar_diff: int=1):
        if config.statsmodels:
            try:
                res = coint_johansen(data, det_order, k_ar_diff)
                return {'eig': res.eig.tolist(), 'lr1': res.lr1.tolist(), 'cvm': res.cvt.tolist()}
            except Exception as e:
                logger.warning(f"Johansen failed: {e}")
                return None
        else:
            return None

class PortfolioOptimization:
    """Black-Litterman and Markowitz using cvxpy if available, else SciPy minimize"""

    @staticmethod
    def gmv_weights(returns: np.ndarray):
        R = np.asarray(returns)
        cov = np.cov(R.T)
        n = cov.shape[0]
        if CVXPY_AVAILABLE:
            w = cp.Variable(n)
            prob = cp.Problem(cp.Minimize(cp.quad_form(w, cov)),
                              [cp.sum(w) == 1])
            prob.solve(solver=cp.SCS, verbose=False)
            w_opt = np.array(w.value).ravel()
            return w_opt
        else:
            # analytic GMV: invcov * 1 / (1^T invcov 1)
            invcov = np.linalg.pinv(cov)
            ones = np.ones((n,))
            w = invcov.dot(ones)
            w = w / (ones.dot(invcov).dot(ones))
            return w

    @staticmethod
    def mean_variance_opt(returns: np.ndarray, target_return: Optional[float]=None):
        R = np.asarray(returns)
        mu = np.mean(R, axis=0)
        cov = np.cov(R.T)
        n = len(mu)
        if CVXPY_AVAILABLE:
            w = cp.Variable(n)
            constraints = [cp.sum(w) == 1]
            if target_return is not None:
                constraints.append(mu @ w >= target_return)
            prob = cp.Problem(cp.Minimize(cp.quad_form(w, cov)), constraints)
            prob.solve(solver=cp.SCS, verbose=False)
            return np.array(w.value).ravel()
        else:
            # solve using analytical formula for target_return or GMV fallback
            if target_return is None:
                return PortfolioOptimization.gmv_weights(R)
            invcov = np.linalg.pinv(cov)
            ones = np.ones(n)
            A = ones.T.dot(invcov).dot(ones)
            B = ones.T.dot(invcov).dot(mu)
            C = mu.T.dot(invcov).dot(mu)
            denom = A * C - B**2
            lam = (C - target_return * B) / denom
            gamma = (target_return * A - B) / denom
            w = invcov.dot(lam * ones + gamma * mu)
            return w

# ---------------------
# ML for Finance helpers
# ---------------------
class MLForFinance:
    @staticmethod
    def compute_basic_features(price: np.ndarray, mom_window: int=20, vol_window: int=20):
        p = np.asarray(price).ravel()
        ret = np.concatenate([[0], np.diff(np.log(p + 1e-12))])
        mom = pd.Series(p).pct_change(mom_window).fillna(0).values
        rv = pd.Series(ret).rolling(vol_window).std().fillna(method='bfill').values
        sma = pd.Series(p).rolling(mom_window).mean().fillna(method='bfill').values
        features = np.vstack([ret, mom, rv, sma]).T
        return features

    @staticmethod
    def lasso_select(X: np.ndarray, y: np.ndarray):
        model = None
        try:
            from sklearn.linear_model import LassoCV
            model = LassoCV(cv=5, n_jobs=1).fit(X, y.ravel())
            coef = model.coef_
            selected = list(np.where(np.abs(coef) > 1e-6)[0])
            return {'coef': coef.tolist(), 'selected': selected, 'alpha': float(model.alpha_)}
        except Exception as e:
            logger.warning(f"LASSO selection failed: {e}")
            return None

# ---------------------
# LLM interface (detailed prompt + structured JSON output)
# ---------------------
class LLMInterface:
    def __init__(self, model_name: str = None, hf_token: Optional[str] = None):
        self.model_name = model_name or config.hf_default_model
        self.api_url = f"https://api-inference.huggingface.co/models/{self.model_name}"
        self.hf_token = hf_token or config.hf_token

    def _call_api(self, prompt: str, max_length: int = 700) -> str:
        headers = {"Authorization": f"Bearer {self.hf_token}"} if self.hf_token else {"Content-Type": "application/json"}
        payload = {"inputs": prompt, "parameters": {"max_new_tokens": max_length, "temperature": 0.2}}
        try:
            r = requests.post(self.api_url, headers=headers, json=payload, timeout=40)
            if r.status_code == 200:
                res = r.json()
                if isinstance(res, list) and isinstance(res[0], dict):
                    return res[0].get("generated_text", str(res[0]))
                return str(res)
            else:
                return f"API_ERROR_{r.status_code}: {r.text[:200]}"
        except Exception as e:
            return f"API_EXCEPTION: {e}"

    def generate_structured_strategy(self,
                                     analysis: Dict[str, Any],
                                     market_snapshot: str,
                                     requirements: Dict[str, Any]) -> Dict[str, Any]:
        """
        Produce structured JSON with keys:
        - strategy_summary
        - signals (list of rules)
        - risk_management
        - pseudocode (string)
        - backtest_guidance
        """
        instr = (
            "You are a quantitative researcher writing a concise Quant Research Note. "
            "Produce structured JSON only, with keys: strategy_summary, signals, risk_management, pseudocode, backtest_guidance, notes.\n\n"
            "Requirements: "
            f"{json.dumps(requirements)}\n\n"
            "Analysis (numerical results):\n"
            f"{json.dumps(analysis, indent=2, ensure_ascii=False)[:4000]}\n\n"
            "Market snapshot:\n"
            f"{market_snapshot[:2000]}\n\n"
            "Be specific: signals should include exact mathematical conditions (e.g. vrp > vrp_sma_short AND rsi < 30). "
            "Pseudocode should include function signatures: compute_features(data), generate_signal(features), risk_manage(position), execute(signal). "
            "Backtest guidance should specify data frequency, in-sample/out-of-sample split, sample length, and slippage/commission assumptions. "
            "Keep outputs compact but precise."
        )
        raw = self._call_api(instr, max_length=800)
        # Try to parse JSON from raw; if fails, fallback to heuristics
        try:
            # sometimes HF returns text with JSON in it — try to extract first JSON object
            start = raw.find("{")
            end = raw.rfind("}")
            if start != -1 and end != -1:
                candidate = raw[start:end+1]
                data = json.loads(candidate)
                return data
        except Exception as e:
            logger.warning(f"LLM did not return pure JSON: {e}")
        # fallback: craft deterministic template using analysis and requirements
        fallback = {
            "strategy_summary": "Fallback strategy: VRP mean-reversion with momentum filter.",
            "signals": [
                "entry: vrp < vrp_sma_short and momentum > 0.5",
                "exit: vrp > vrp_sma_long or price crosses stop loss"
            ],
            "risk_management": "max position risk 0.5% NAV; use stop-loss and time-based exit",
            "pseudocode": (
                "def compute_features(data):\n"
                "    features = {...}  # vrp, sma, momentum\n"
                "def generate_signal(features):\n"
                "    if features['vrp'] < features['vrp_sma_short'] and features['mom'] > 0:\n"
                "        return 1\n"
                "    return 0\n"
                "def risk_manage(pos):\n"
                "    # apply stop loss / position sizing\n"
            ),
            "backtest_guidance": "Use 1-minute bars, in-sample 2 years, OOS 6 months, slippage 0.02%, commission 0.0005 per trade",
            "notes": "LLM API failed or returned non-JSON; this is a deterministic fallback."
        }
        return fallback

# ---------------------
# Integrative Trainer / Platform
# ---------------------
class QuantPlatform:
    def __init__(self):
        self.fiber = FiberBundleTheory()
        self.noise = NoiseExplorer()
        self.trainer_ml = None
        self.llm = LLMInterface()
        self.current_data = None
        self.analysis_results = {}

    # Data ingestion & basic analysis
    def upload_and_analyze(self, file):
        if file is None:
            return "请上传 CSV / Excel 文件", None, None
        fname = file.name
        try:
            if fname.endswith('.csv'):
                df = pd.read_csv(fname)
            else:
                df = pd.read_excel(fname)
        except Exception as e:
            return f"读取失败: {e}", None, None
        self.current_data = df
        numeric = df.select_dtypes(include=[np.number]).columns.tolist()
        summary = f"Rows: {len(df)}, Cols: {len(df.columns)}, Numeric: {numeric}"
        # noise exploration (first two numeric columns)
        try:
            noise_res = self.noise.explore(df)
            noise_summary = f"VRP mean {noise_res['vrp_mean']:.6f}, vrp std {noise_res['vrp_std']:.6f}, resid ac1 {noise_res['resid_stats']['ac1']:.4f}"
        except Exception as e:
            noise_summary = f"噪声分析失败: {e}"
            noise_res = None
        # garch quick fit on first numeric column returns (if plausible)
        garch_summary = "GARCH not run"
        if numeric:
            series = df[numeric[0]].pct_change().dropna().values
            if len(series) > 30:
                try:
                    garch_res = Econometrics.garch_11_fit(series)
                    garch_summary = f"GARCH method: {garch_res.get('method','?')}, params keys: {list(garch_res.get('params',{}).keys()) if 'params' in garch_res else 'n/a'}"
                except Exception as e:
                    garch_summary = f"GARCH失败: {e}"
        self.analysis_results = {'noise': noise_res, 'garch': garch_summary}
        return summary, noise_summary, garch_summary

    # Pricing / PDE / MC wrappers
    def price_bs_cn(self, S, K, r, q, sigma, T, Smax_mult=3.0, M=400, N=400, option_type='call'):
        try:
            p = NumericalMethods.bs_crank_nicolson(float(S), float(K), float(r), float(q), float(sigma), float(T),
                                                  Smax_mult=float(Smax_mult), M=int(M), N=int(N), option_type=option_type)
            return f"Crank–Nicolson price: {p:.6f}"
        except Exception as e:
            return f"PDE pricing failed: {e}"

    def price_bs_mc(self, S, K, r, q, sigma, T, option_type='call', n_paths=20000, antithetic=True):
        try:
            p = NumericalMethods.mc_price_bs_cv(float(S), float(K), float(r), float(q), float(sigma), float(T),
                                               option_type=option_type, n_paths=int(n_paths), antithetic=bool(antithetic))
            return f"MC price (CV): {p:.6f}"
        except Exception as e:
            return f"MC pricing failed: {e}"

    def simulate_heston(self, S0, v0, r, kappa, theta, xi, rho, T, n_steps=252, n_paths=2000):
        try:
            S, v = StochasticModels.heston_simulate(float(S0), float(v0), float(r), float(kappa), float(theta), float(xi), float(rho), float(T), int(n_steps), int(n_paths))
            # return minimal summary and a small plot (first 3 paths)
            fig, ax = plt.subplots()
            for i in range(min(3, S.shape[0])):
                ax.plot(S[i,:], label=f'path{i}')
            ax.set_title("Heston sample paths (first few)")
            ax.legend()
            return "Heston simulation success", fig
        except Exception as e:
            return f"Heston simulation failed: {e}", None

    # Econometrics wrappers
    def garch_fit(self):
        if self.current_data is None:
            return "请先上传数据"
        numeric = self.current_data.select_dtypes(include=[np.number]).columns.tolist()
        if not numeric:
            return "数据无数值列"
        series = self.current_data[numeric[0]].pct_change().dropna().values
        if len(series) < 30:
            return "样本过短，至少需要30个观测用于GARCH拟合"
        try:
            res = Econometrics.garch_11_fit(series)
            return json.dumps({'method': res.get('method','mle'), 'params': res.get('params') if 'params' in res else 'omega/alpha/beta', 'cond_var_mean': float(np.mean(res.get('cond_var',[])) if res.get('cond_var') else np.nan)}, indent=2)
        except Exception as e:
            return f"GARCH拟合失败: {e}"

    def johansen(self):
        if self.current_data is None:
            return "请先上传数据"
        data = self.current_data.select_dtypes(include=[np.number]).dropna().values
        if data.shape[0] < 50 or data.shape[1] < 2:
            return "数据不足以做 Johansen 协整检验（至少 50 行，2 列）"
        try:
            res = Econometrics.johansen_test(data)
            if res is None:
                return "Johansen 不可用（statsmodels 未安装或出错）"
            return json.dumps({'eig_top5': res['eig'][:5], 'lr1_top5': res['lr1'][:5]}, indent=2)
        except Exception as e:
            return f"Johansen 失败: {e}"

    # Portfolio & Risk
    def compute_gmv(self):
        if self.current_data is None:
            return "请先上传数据"
        df = self.current_data.select_dtypes(include=[np.number]).dropna()
        if df.shape[0] < 10 or df.shape[1] < 1:
            return "数据不足"
        returns = df.pct_change().dropna().values
        w = PortfolioOptimization.gmv_weights(returns)
        return f"GMV weights (len {len(w)}): {np.round(w,4).tolist()}"

    def mean_var_opt(self, target_return: Optional[float]=None):
        if self.current_data is None:
            return "请先上传数据"
        df = self.current_data.select_dtypes(include=[np.number]).dropna()
        returns = df.pct_change().dropna().values
        try:
            w = PortfolioOptimization.mean_variance_opt(returns, target_return=float(target_return) if target_return is not None else None)
            return f"Optimized weights (len {len(w)}): {np.round(w,4).tolist()}"
        except Exception as e:
            return f"Mean-Variance optimization failed: {e}"

    # ML
    def lasso_select(self):
        if self.current_data is None:
            return "请先上传数据"
        df = self.current_data.select_dtypes(include=[np.number]).dropna()
        if df.shape[1] < 2 or df.shape[0] < 30:
            return "数据不足以做 LASSO"
        y = df.iloc[:,0].pct_change().dropna().values
        X = df.iloc[:,1:].pct_change().dropna().values
        # align lengths
        minlen = min(len(y), len(X))
        if minlen <= 10:
            return "数据对齐后样本太短"
        y = y[-minlen:]
        X = X[-minlen:]
        res = MLForFinance.lasso_select(X, y)
        if res is None:
            return "LASSO 失败"
        return f"Selected indices: {res['selected']}, alpha: {res['alpha']:.6g}"

    # LLM strategy (structured)
    def generate_strategy(self, user_prompt: str, intraday: bool=True, model_name: Optional[str]=None) -> str:
        if self.current_data is None:
            return json.dumps({'error': '请先上传数据'}, ensure_ascii=False)
        # Build analysis dict
        analysis = {}
        if self.analysis_results.get('noise'):
            analysis['noise'] = self.analysis_results['noise']
        # GARCH cond var mean if available
        try:
            g = self.garch_fit()
            analysis['garch_summary'] = json.loads(g) if g and g.startswith("{") else g
        except Exception:
            analysis['garch_summary'] = "GARCH无法解析"
        # market snapshot: last 50 rows numeric describe
        do_numeric = self.current_data.select_dtypes(include=[np.number]).tail(50).describe().to_string()
        requirements = {'intraday': intraday, 'pseudocode': True, 'user_prompt': user_prompt}
        if model_name:
            self.llm = LLMInterface(model_name=model_name)
        result = self.llm.generate_structured_strategy(analysis, do_numeric, requirements)
        # return pretty JSON
        return json.dumps(result, ensure_ascii=False, indent=2)

# ---------------------
# Gradio UI
# ---------------------
def create_ui():
    platform = QuantPlatform()
    with gr.Blocks(title="Quant Upgraded Platform") as demo:
        gr.Markdown("# Quant Upgraded Platform — 高精度/高性能 + 精细化 LLM 策略")
        with gr.Tabs():
            with gr.TabItem("📁 数据上传 & 基础分析"):
                with gr.Row():
                    file_input = gr.File(label="上传 CSV / Excel")
                    upload_btn = gr.Button("上传并分析")
                summary = gr.Textbox(label="数据摘要", lines=2)
                noise = gr.Textbox(label="噪声探索摘要", lines=2)
                garch = gr.Textbox(label="GARCH 摘要", lines=2)
                upload_btn.click(platform.upload_and_analyze, inputs=[file_input], outputs=[summary, noise, garch])

            with gr.TabItem("📊 Pricing / PDE / MC"):
                with gr.Row():
                    S = gr.Number(value=100.0, label="Spot S")
                    K = gr.Number(value=100.0, label="Strike K")
                    r = gr.Number(value=0.01, label="r")
                    q = gr.Number(value=0.0, label="q")
                    sigma = gr.Number(value=0.2, label="sigma")
                    T = gr.Number(value=0.5, label="T (yrs)")
                with gr.Row():
                    bs_cn_btn = gr.Button("Crank–Nicolson BS PDE 价格")
                    bs_cn_out = gr.Textbox(label="PDE Price", lines=1)
                    bs_cn_btn.click(platform.price_bs_cn, inputs=[S,K,r,q,sigma,T, gr.Number(value=3.0), gr.Slider(100,800,value=400), gr.Slider(100,800,value=400), gr.Dropdown(['call','put'], value='call')], outputs=[bs_cn_out])
                with gr.Row():
                    mc_btn = gr.Button("Monte Carlo (Antithetic + Control Var)")
                    mc_out = gr.Textbox(label="MC Price (CV)", lines=1)
                    mc_btn.click(platform.price_bs_mc, inputs=[S,K,r,q,sigma,T, gr.Dropdown(['call','put'], value='call'), gr.Number(value=config.mc_default_paths), gr.Checkbox(value=True, label="Antithetic")], outputs=[mc_out])

            with gr.TabItem("🔢 Econometrics"):
                garch_btn = gr.Button("GARCH(1,1) 拟合")
                garch_out = gr.Textbox(label="GARCH 结果", lines=8)
                garch_btn.click(platform.garch_fit, inputs=None, outputs=[garch_out])

                joh_btn = gr.Button("Johansen 协整检验")
                joh_out = gr.Textbox(label="Johansen 结果", lines=6)
                joh_btn.click(platform.johansen, inputs=None, outputs=[joh_out])

            with gr.TabItem("📈 Portfolio & Risk"):
                gmv_btn = gr.Button("计算 GMV 权重")
                gmv_out = gr.Textbox(label="GMV 权重", lines=3)
                gmv_btn.click(platform.compute_gmv, inputs=None, outputs=[gmv_out])

                mv_btn = gr.Button("均值-方差 优化 (可选目标收益)")
                target = gr.Number(label="目标收益 (可空)", value=None)
                mv_out = gr.Textbox(label="MV 结果", lines=3)
                mv_btn.click(platform.mean_var_opt, inputs=[target], outputs=[mv_out])

            with gr.TabItem("🤖 LLM 策略生成 (结构化)"):
                user_q = gr.Textbox(label="你的问题（策略 / 日内 / 回测）", lines=3, value="基于当前数据，给出日内量化策略并生成伪代码")
                intraday = gr.Checkbox(label="日内策略", value=True)
                model_sel = gr.Dropdown(label="LLM 模型 (若无Token或模型不可用会回退)", choices=[config.hf_default_model], value=config.hf_default_model)
                strat_out = gr.Textbox(label="结构化策略输出 (JSON)", lines=20)
                strat_btn = gr.Button("生成策略")
                strat_btn.click(platform.generate_strategy, inputs=[user_q, intraday, model_sel], outputs=[strat_out])

            with gr.TabItem("🔬 Dynamics & Geometry (原有)"):
                noise_btn = gr.Button("运行噪声探索")
                noise_text = gr.Textbox(label="Noise summary", lines=3)
                def run_noise():
                    if platform.current_data is None:
                        return "请先上传数据"
                    res = platform.noise.explore(platform.current_data)
                    return f"VRP mean {res['vrp_mean']:.6f}, resid ac1 {res['resid_stats']['ac1']:.4f}"
                noise_btn.click(run_noise, inputs=None, outputs=[noise_text])

                sim_vix2 = gr.Number(value=1.0, label="start VIX^2")
                sim_rv = gr.Number(value=0.8, label="start RV")
                T_sim = gr.Number(value=1.0, label="T")
                dt_sim = gr.Number(value=0.01, label="dt")
                sim_btn = gr.Button("模拟梯度动力学")
                sim_out = gr.Plot(label="Dynamics path")
                def run_sim(vix2, rv, T, dt):
                    # lightweight simulate using gradient dynamics (reuse earlier pattern)
                    gradient = GradientDynamicsLite()
                    path = gradient.simulate_flow([vix2, rv], T=float(T), dt=float(dt))
                    fig, ax = plt.subplots()
                    ax.plot(path[:,0], label='VIX^2')
                    ax.plot(path[:,1], label='RV')
                    ax.legend()
                    ax.set_title("Gradient dynamics (VIX^2 & RV)")
                    return fig
                sim_btn.click(run_sim, inputs=[sim_vix2, sim_rv, T_sim, dt_sim], outputs=[sim_out])

        gr.Markdown("注：本系统为研究用途，不构成投资建议。部分功能依赖外部库（statsmodels, arch, cvxpy）。")

    return demo

# ---------------------
# Small helper: GradientDynamicsLite (used only in UI simulation)
# ---------------------
class GradientDynamicsLite:
    def __init__(self, eta=0.5, sigma=0.02):
        self.eta = eta
        self.sigma = sigma

    def U_vrp(self, b):
        vix2 = b[...,0]
        rv = b[...,1]
        vrp = vix2 - rv
        return 0.5 * vrp**2

    def grad_U(self, b):
        # analytic gradient for U = 0.5*(vix2 - rv)^2
        vix2 = b[0]
        rv = b[1]
        # dU/dvix2 = (vix2 - rv); dU/drv = -(vix2 - rv)
        g = np.array([vix2 - rv, -(vix2 - rv)], dtype=float)
        return g

    def simulate_flow(self, b0, T=1.0, dt=0.01, seed=None):
        if seed is not None:
            np.random.seed(seed)
        n_steps = int(T / dt)
        path = np.zeros((n_steps+1, 2))
        path[0] = np.array(b0, dtype=float)
        for i in range(n_steps):
            bcur = path[i]
            grad = self.grad_U(bcur)
            db_det = - self.eta * grad
            db_stoch = self.sigma * np.sqrt(dt) * np.random.randn(2)
            path[i+1] = bcur + db_det * dt + db_stoch
        return path

# ---------------------
# Entrypoint
# ---------------------
if __name__ == "__main__":
    app = create_ui()
    # Launch locally
    app.launch(server_name="0.0.0.0", server_port=7860, share=False)