Spaces:
Sleeping
Sleeping
| import logging | |
| import sys | |
| import time | |
| from typing import List, Optional | |
| import gradio as gr | |
| from config import ( | |
| DEFAULT_SYMBOLS, | |
| TOP_N_DEFAULT, | |
| DEFAULT_ACCOUNT_EQUITY, | |
| TIMEFRAME, | |
| CANDLE_LIMIT, | |
| ) | |
| from data_fetcher import fetch_multiple, fetch_instruments | |
| from regime import detect_regime | |
| from volume_analysis import analyze_volume | |
| from risk_engine import evaluate_risk | |
| from veto import apply_veto, veto_summary | |
| from scorer import compute_structure_score, score_token, rank_tokens, format_score_bar | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", | |
| stream=sys.stdout, | |
| ) | |
| logger = logging.getLogger("main") | |
| _TREND_EMOJI = {"bullish": "🟢", "ranging": "🟡", "bearish": "🔴"} | |
| _BREAKOUT_LABEL = {1: "↑ UP", -1: "↓ DOWN", 0: "—"} | |
| def analyze_single(symbol: str, df, account_equity: float) -> dict: | |
| regime_data = detect_regime(df) | |
| volume_data = analyze_volume(df) | |
| structure_score = compute_structure_score(regime_data) | |
| vetoed, veto_reason = apply_veto(regime_data, volume_data, structure_score) | |
| scores = score_token(regime_data, volume_data, vetoed) | |
| risk_data = evaluate_risk( | |
| close=float(df["close"].iloc[-1]), | |
| atr=regime_data["atr"], | |
| atr_pct=regime_data["atr_pct"], | |
| regime_score=regime_data["regime_score"], | |
| vol_ratio=regime_data["vol_ratio"], | |
| volume_score=volume_data["volume_score"], | |
| account_equity=account_equity, | |
| ) | |
| return { | |
| "symbol": symbol, | |
| "close": float(df["close"].iloc[-1]), | |
| "trend": regime_data["trend"], | |
| "vol_ratio": regime_data["vol_ratio"], | |
| "vol_expanding": regime_data["vol_expanding"], | |
| "atr_pct": regime_data["atr_pct"], | |
| "spike": volume_data["spike"], | |
| "climax": volume_data["climax"], | |
| "breakout": volume_data["breakout"], | |
| "obv_slope": volume_data["obv_slope_norm"], | |
| "delta_sign": volume_data["delta_sign"], | |
| "vetoed": vetoed, | |
| "veto_reason": veto_reason, | |
| "regime_score": scores["regime_score"], | |
| "volume_score": scores["volume_score"], | |
| "structure_score": scores["structure_score"], | |
| "total_score": scores["total_score"], | |
| "risk": risk_data, | |
| } | |
| def build_summary_table(ranked: list, top_n: int) -> str: | |
| header = ( | |
| f"{'#':>3} {'Symbol':<14} {'Score':>6} {'Regime':>7} " | |
| f"{'Volume':>7} {'Structure':>10} {'Trend':<8} " | |
| f"{'VolRatio':>8} {'Spike':>5} {'BOS':>5} {'Status'}\n" | |
| ) | |
| separator = "─" * 110 + "\n" | |
| rows = header + separator | |
| for rank, (sym, data) in enumerate(ranked[:top_n], 1): | |
| trend_icon = _TREND_EMOJI.get(data["trend"], "⚪") | |
| breakout_lbl = _BREAKOUT_LABEL.get(data["breakout"], "—") | |
| spike_lbl = "✓" if data["spike"] else "✗" | |
| status = "VETOED" if data["vetoed"] else "OK" | |
| rows += ( | |
| f"{rank:>3} {sym:<14} {data['total_score']:>6.4f} " | |
| f"{data['regime_score']:>7.4f} {data['volume_score']:>7.4f} " | |
| f"{data['structure_score']:>10.4f} " | |
| f"{trend_icon} {data['trend']:<6} " | |
| f"{data['vol_ratio']:>8.2f} {spike_lbl:>5} " | |
| f"{breakout_lbl:>5} {status}\n" | |
| ) | |
| return rows | |
| def build_top_setup_detail(data: dict) -> str: | |
| r = data["risk"] | |
| sym = data["symbol"] | |
| trend_icon = _TREND_EMOJI.get(data["trend"], "⚪") | |
| lines = [ | |
| "═" * 60, | |
| f" BEST SETUP: {sym}", | |
| "═" * 60, | |
| f" Trend: {trend_icon} {data['trend'].upper()}", | |
| f" Close Price: {r['entry_price']:.8f}", | |
| f"", | |
| f" ── SCORES ──────────────────────────────", | |
| f" Regime: {format_score_bar(data['regime_score'])}", | |
| f" Volume: {format_score_bar(data['volume_score'])}", | |
| f" Structure: {format_score_bar(data['structure_score'])}", | |
| f" Total: {format_score_bar(data['total_score'])}", | |
| f"", | |
| f" ── RISK PARAMETERS ──────────────────────", | |
| f" ATR: {r['atr']:.8f} ({r['atr_pct']:.3f}%)", | |
| f" Vol Ratio: {r['vol_ratio']:.2f}x (quality: {r['risk_quality']:.0%})", | |
| f" Risk Fraction: {r['risk_fraction']:.3f}%", | |
| f" $ At Risk: ${r['dollar_at_risk']:.2f}", | |
| f" Position Size: ${r['position_notional']:.2f} notional", | |
| f" Leverage (est): {r['leverage_implied']:.1f}x", | |
| f"", | |
| f" ── LONG SCENARIO ───────────────────────", | |
| f" Stop Loss: {r['stop_long']:.8f}", | |
| f" Take Profit: {r['target_long']:.8f}", | |
| f" R:R Ratio: 1 : {r['rr_ratio']:.1f}", | |
| f"", | |
| f" ── SHORT SCENARIO ──────────────────────", | |
| f" Stop Loss: {r['stop_short']:.8f}", | |
| f" Take Profit: {r['target_short']:.8f}", | |
| f" R:R Ratio: 1 : {r['rr_ratio']:.1f}", | |
| f"", | |
| f" Volume Spike: {'YES ✓' if data['spike'] else 'NO ✗'}", | |
| f" Volume Climax: {'YES ⚠' if data['climax'] else 'NO'}", | |
| f" Breakout: {_BREAKOUT_LABEL.get(data['breakout'], '—')}", | |
| f" OBV Slope: {data['obv_slope']:+.4f}", | |
| f" Delta (5-bar): {'BUYING ↑' if data['delta_sign'] > 0 else 'SELLING ↓'}", | |
| "═" * 60, | |
| ] | |
| return "\n".join(lines) | |
| def parse_symbol_list(raw: str) -> List[str]: | |
| symbols = [] | |
| for tok in raw.replace(",", " ").replace("\n", " ").split(): | |
| tok = tok.strip().upper() | |
| if tok and "-" in tok: | |
| symbols.append(tok) | |
| elif tok: | |
| symbols.append(f"{tok}-USDT") | |
| return symbols if symbols else DEFAULT_SYMBOLS | |
| def run_analysis( | |
| symbols_input: str, | |
| equity: float, | |
| top_n: int, | |
| use_live_instruments: bool, | |
| progress=gr.Progress(track_tqdm=False), | |
| ) -> str: | |
| start_ts = time.time() | |
| output_lines = [] | |
| output_lines.append("━" * 60) | |
| output_lines.append(" OKX QUANTITATIVE ANALYSIS ENGINE") | |
| output_lines.append("━" * 60) | |
| if use_live_instruments: | |
| output_lines.append("⟳ Fetching live instrument list from OKX...") | |
| symbols = fetch_instruments("SPOT") | |
| if not symbols: | |
| output_lines.append("⚠ Failed to fetch live instruments. Using defaults.") | |
| symbols = DEFAULT_SYMBOLS | |
| else: | |
| output_lines.append(f"✓ Found {len(symbols)} live USDT spot instruments") | |
| else: | |
| symbols = parse_symbol_list(symbols_input) | |
| output_lines.append(f"✓ Analyzing {len(symbols)} symbol(s)") | |
| output_lines.append(f" Timeframe: {TIMEFRAME} | Candles: {CANDLE_LIMIT} | Equity: ${equity:,.0f}") | |
| output_lines.append("") | |
| fetched_count = [0] | |
| total = len(symbols) | |
| def progress_cb(i, t, sym): | |
| fetched_count[0] = i | |
| progress(i / t, desc=f"Fetching {sym} ({i}/{t})") | |
| ohlcv_map = fetch_multiple( | |
| symbols, | |
| min_bars=50, | |
| progress_callback=progress_cb, | |
| ) | |
| output_lines.append(f"✓ Fetched {len(ohlcv_map)}/{total} symbols successfully") | |
| output_lines.append("") | |
| all_results = {} | |
| failed = [] | |
| for sym, df in ohlcv_map.items(): | |
| try: | |
| all_results[sym] = analyze_single(sym, df, account_equity=equity) | |
| except Exception as exc: | |
| logger.error(f"Analysis error for {sym}: {exc}", exc_info=True) | |
| failed.append(sym) | |
| if failed: | |
| output_lines.append(f"⚠ Analysis failed for: {', '.join(failed)}") | |
| output_lines.append("") | |
| ranked = rank_tokens(all_results) | |
| approved = [(s, d) for s, d in ranked if not d["vetoed"]] | |
| vetoed_count = sum(1 for _, d in ranked if d["vetoed"]) | |
| output_lines.append(f" RESULTS: {len(all_results)} analyzed | {len(approved)} approved | {vetoed_count} vetoed") | |
| output_lines.append("") | |
| output_lines.append(" TOP SETUPS RANKED BY TOTAL SCORE") | |
| output_lines.append("─" * 110) | |
| output_lines.append(build_summary_table(ranked, int(top_n))) | |
| if approved: | |
| best_sym, best_data = approved[0] | |
| output_lines.append("") | |
| output_lines.append(build_top_setup_detail(best_data)) | |
| else: | |
| output_lines.append("") | |
| output_lines.append(" ⚠ No approved setups found — all tokens vetoed.") | |
| elapsed = time.time() - start_ts | |
| output_lines.append("") | |
| output_lines.append(f" ✓ Analysis complete in {elapsed:.1f}s") | |
| output_lines.append("━" * 60) | |
| return "\n".join(output_lines) | |
| def build_interface() -> gr.Blocks: | |
| with gr.Blocks( | |
| title="OKX Quant Analysis Engine", | |
| theme=gr.themes.Base( | |
| primary_hue="slate", | |
| neutral_hue="slate", | |
| font=[gr.themes.GoogleFont("JetBrains Mono"), "monospace"], | |
| ), | |
| css=""" | |
| body { background: #0a0a0f; } | |
| .gradio-container { | |
| background: #0a0a0f !important; | |
| max-width: 1100px !important; | |
| font-family: 'JetBrains Mono', monospace !important; | |
| } | |
| .gr-button-primary { | |
| background: #1a6bff !important; | |
| border: none !important; | |
| font-family: 'JetBrains Mono', monospace !important; | |
| font-weight: 700 !important; | |
| letter-spacing: 0.05em !important; | |
| } | |
| .gr-button-primary:hover { background: #0050e0 !important; } | |
| #output_box textarea { | |
| font-family: 'JetBrains Mono', monospace !important; | |
| font-size: 13px !important; | |
| background: #0f0f1a !important; | |
| color: #c8d0e0 !important; | |
| border: 1px solid #1e2236 !important; | |
| min-height: 700px !important; | |
| } | |
| label, .gr-form label { | |
| font-family: 'JetBrains Mono', monospace !important; | |
| color: #8899bb !important; | |
| font-size: 11px !important; | |
| letter-spacing: 0.08em !important; | |
| text-transform: uppercase !important; | |
| } | |
| .gr-panel { background: #0d0d18 !important; border: 1px solid #1e2236 !important; } | |
| h1 { color: #e0e8ff !important; font-family: 'JetBrains Mono', monospace !important; letter-spacing: 0.05em !important; } | |
| p { color: #5a6a8a !important; font-family: 'JetBrains Mono', monospace !important; font-size: 12px !important; } | |
| """, | |
| ) as app: | |
| gr.Markdown("# ◈ OKX QUANT ANALYSIS ENGINE") | |
| gr.Markdown("Multi-token regime detection · volume analysis · dynamic risk · veto scoring") | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| symbols_box = gr.Textbox( | |
| label="Symbols (comma or newline separated — leave blank for defaults)", | |
| placeholder="BTC-USDT, ETH-USDT, SOL-USDT ...", | |
| lines=4, | |
| value="", | |
| ) | |
| with gr.Column(scale=1): | |
| equity_slider = gr.Slider( | |
| label="Account Equity (USD)", | |
| minimum=100, | |
| maximum=1_000_000, | |
| step=100, | |
| value=DEFAULT_ACCOUNT_EQUITY, | |
| ) | |
| top_n_slider = gr.Slider( | |
| label="Top N Results to Show", | |
| minimum=5, | |
| maximum=100, | |
| step=5, | |
| value=TOP_N_DEFAULT, | |
| ) | |
| live_instruments = gr.Checkbox( | |
| label="Fetch live instruments from OKX (100+ symbols)", | |
| value=False, | |
| ) | |
| run_btn = gr.Button("▶ RUN ANALYSIS", variant="primary", size="lg") | |
| output_box = gr.Textbox( | |
| label="Analysis Output", | |
| lines=40, | |
| max_lines=100, | |
| interactive=False, | |
| elem_id="output_box", | |
| ) | |
| run_btn.click( | |
| fn=run_analysis, | |
| inputs=[symbols_box, equity_slider, top_n_slider, live_instruments], | |
| outputs=output_box, | |
| ) | |
| gr.Markdown( | |
| "**Signals are for research purposes only. Not financial advice.** " | |
| "Data sourced from OKX public REST API." | |
| ) | |
| return app | |
| if __name__ == "__main__": | |
| app = build_interface() | |
| app.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| show_error=True, | |
| share=False, | |
| ) | |