""" FASE 4 — Evaluation Script Test semua perubahan FASE 4 dengan data sintetik: 1. Market Microstructure: klasifikasi kombinasi volume + price 2. Foreign Flow Predictivity: korelasi → weight adjustment 3. Earnings Recency: recency label & weight 4. Ensemble Weighted Blend: kombinasi 5 sinyal Output: backend/data/phase4_evaluation.json """ import sys import os import json from datetime import datetime, timedelta from pathlib import Path # Add project root to path project_root = Path(__file__).resolve().parent.parent sys.path.insert(0, str(project_root)) import numpy as np import pandas as pd def test_market_microstructure(): """Test 1: Klasifikasi mikrostruktur volume + price.""" print("\n" + "=" * 60) print("TEST 1: Market Microstructure Classification") print("=" * 60) from app.analytics.market_microstructure import classify_microstructure results = [] # Scenario A: Silent Accumulation (Vol HIGH + Price FLAT) np.random.seed(42) n = 25 dates = pd.date_range(end=datetime.now(), periods=n) prices_flat = pd.DataFrame({ "open": np.random.uniform(99, 101, n), "high": np.random.uniform(100, 102, n), "low": np.random.uniform(98, 100, n), "close": np.concatenate([np.random.uniform(99, 101, n-1), [100.5]]), "volume": np.concatenate([np.random.uniform(1e6, 2e6, n-1), [8e6]]), }, index=dates) vol_series = prices_flat["volume"] sig_a = classify_microstructure(prices_flat, vol_series) print(f" A. Silent Accumulation: {sig_a.category} ({sig_a.direction})") print(f" Confidence: {sig_a.confidence}, Weight: {sig_a.ensemble_weight}") results.append({ "scenario": "silent_accumulation", "expected": "SILENT_ACCUMULATION", "actual": sig_a.category, "pass": sig_a.category == "SILENT_ACCUMULATION", }) # Scenario B: Breakout Momentum (Vol HIGH + Price UP) prices_breakout = prices_flat.copy() prices_breakout.loc[prices_breakout.index[-1], "close"] = 106.0 # +6% prices_breakout.loc[prices_breakout.index[-1], "open"] = 100.0 prices_breakout.loc[prices_breakout.index[-2], "close"] = 100.0 sig_b = classify_microstructure(prices_breakout, vol_series) print(f" B. Breakout: {sig_b.category} ({sig_b.direction})") print(f" Confidence: {sig_b.confidence}, Weight: {sig_b.ensemble_weight}") results.append({ "scenario": "breakout_momentum", "expected": "BREAKOUT_MOMENTUM", "actual": sig_b.category, "pass": sig_b.category == "BREAKOUT_MOMENTUM", }) # Scenario C: Panic Sell (Vol HIGH + Price DROP) prices_panic = prices_flat.copy() prices_panic.loc[prices_panic.index[-1], "close"] = 94.0 # -6% prices_panic.loc[prices_panic.index[-1], "open"] = 100.0 prices_panic.loc[prices_panic.index[-2], "close"] = 100.0 sig_c = classify_microstructure(prices_panic, vol_series) print(f" C. Panic Sell: {sig_c.category} ({sig_c.direction})") print(f" Confidence: {sig_c.confidence}, Weight: {sig_c.ensemble_weight}") results.append({ "scenario": "panic_sell", "expected": "PANIC_SELL", "actual": sig_c.category, "pass": sig_c.category == "PANIC_SELL", }) # Scenario D: Low Confidence Rally (Vol LOW + Price UP) vol_low = pd.Series( np.concatenate([np.random.uniform(1e6, 2e6, n-1), [1.2e6]]), index=dates ) sig_d = classify_microstructure(prices_breakout, vol_low) print(f" D. Low Conf Rally: {sig_d.category} ({sig_d.direction})") print(f" Confidence: {sig_d.confidence}, Weight: {sig_d.ensemble_weight}") results.append({ "scenario": "low_confidence_rally", "expected": "LOW_CONFIDENCE_RALLY", "actual": sig_d.category, "pass": sig_d.category == "LOW_CONFIDENCE_RALLY", }) # Scenario E: Normal (Vol NORMAL + Price FLAT) vol_normal = pd.Series( np.random.uniform(1e6, 1.3e6, n), index=dates ) prices_normal = prices_flat.copy() prices_normal.loc[prices_normal.index[-1], "close"] = 100.3 prices_normal.loc[prices_normal.index[-2], "close"] = 100.0 sig_e = classify_microstructure(prices_normal, vol_normal) print(f" E. Normal: {sig_e.category} ({sig_e.direction})") print(f" Confidence: {sig_e.confidence}, Weight: {sig_e.ensemble_weight}") results.append({ "scenario": "normal", "expected": "NORMAL", "actual": sig_e.category, "pass": sig_e.category == "NORMAL", }) passed = sum(1 for r in results if r["pass"]) print(f"\n Result: {passed}/{len(results)} scenarios passed") return results def test_foreign_flow_predictivity(): """Test 2: Foreign flow predictivity validation.""" print("\n" + "=" * 60) print("TEST 2: Foreign Flow Predictivity Validation") print("=" * 60) from app.analytics.foreign_flow import compute_foreign_flow_predictivity np.random.seed(42) n = 60 dates = pd.date_range(end=datetime.now(), periods=n) # Scenario A: Strong positive correlation (flow predicts price up) flow_strong = pd.Series(np.random.uniform(0.3, 0.8, n), index=dates) returns_strong = flow_strong * 0.05 + np.random.normal(0, 0.01, n) close_strong = 100 * (1 + returns_strong).cumprod() prices_strong = pd.DataFrame({"close": close_strong}, index=dates) pred_a = compute_foreign_flow_predictivity(flow_strong, prices_strong, forward_days=5) print(f" A. Strong positive correlation:") print(f" r = {pred_a['correlation']}, class = {pred_a['predictive_class']}") print(f" weight = {pred_a['ensemble_weight']}, dir = {pred_a['direction']}") # Scenario B: Weak/no correlation flow_weak = pd.Series(np.random.uniform(-0.5, 0.5, n), index=dates) close_weak = pd.Series( 100 * np.cumprod(1 + np.random.normal(0, 0.02, n)), index=dates ) prices_weak = pd.DataFrame({"close": close_weak}, index=dates) pred_b = compute_foreign_flow_predictivity(flow_weak, prices_weak, forward_days=5) print(f" B. Weak correlation:") print(f" r = {pred_b['correlation']}, class = {pred_b['predictive_class']}") print(f" weight = {pred_b['ensemble_weight']}, dir = {pred_b['direction']}") # Scenario C: Insufficient data flow_short = pd.Series([0.5, 0.3], index=dates[:2]) pred_c = compute_foreign_flow_predictivity(flow_short, prices_strong, forward_days=5) print(f" C. Insufficient data:") print(f" class = {pred_c['predictive_class']}, note = {pred_c['note']}") return { "strong": pred_a, "weak": pred_b, "insufficient": pred_c, } def test_earnings_recency(): """Test 3: Earnings recency classification.""" print("\n" + "=" * 60) print("TEST 3: Earnings Recency Classification") print("=" * 60) from app.analytics.earnings_impact import EarningsImpactAnalyzer, EarningsEvent analyzer = EarningsImpactAnalyzer() scenarios = [ ("FRESH_RELEASE", 2), # 2 hari lalu ("RECENT_RELEASE", 5), # 5 hari lalu ("POST_DRIFT", 15), # 15 hari lalu ("STALE", 45), # 45 hari lalu ] results = [] for expected_label, days_ago in scenarios: event = EarningsEvent( symbol="TEST", report_date=datetime.utcnow() - timedelta(days=days_ago), period="Q1", eps_actual=1.10, eps_estimate=1.00, eps_surprise=0.10, eps_surprise_pct=0.10, revenue_actual=None, revenue_estimate=None, impact_label="BEAT", impact_magnitude="MODERATE", days_since_report=days_ago, ) label = analyzer._compute_recency_label(event) weight = analyzer._compute_ensemble_weight(event, label) print(f" {expected_label}: days_ago={days_ago} → {label} (weight={weight})") results.append({ "expected": expected_label, "actual": label, "weight": weight, "pass": label == expected_label, }) passed = sum(1 for r in results if r["pass"]) print(f"\n Result: {passed}/{len(results)} passed") return results def test_ensemble_blend(): """Test 4: Weighted blend ensemble.""" print("\n" + "=" * 60) print("TEST 4: Ensemble Weighted Blend") print("=" * 60) from app.ml.ensemble import EnsemblePredictor, SignalInput predictor = EnsemblePredictor() # Build synthetic OHLCV np.random.seed(42) n = 30 dates = pd.date_range(end=datetime.now(), periods=n) close = pd.Series(100 * np.cumprod(1 + np.random.normal(0.001, 0.02, n)), index=dates) prices = pd.DataFrame({ "open": close * 0.999, "high": close * 1.01, "low": close * 0.99, "close": close, "volume": np.random.uniform(1e6, 3e6, n), }, index=dates) scenarios = [] # Scenario A: All bullish signals sig_bull = SignalInput( ml_score=0.6, ml_available=False, # model belum trained sentiment_score=0.7, sentiment_available=True, foreign_flow_score=0.5, foreign_flow_weight=0.75, # MODERATE predictivity earnings_score=0.4, earnings_weight=1.0, # FRESH_RELEASE microstructure_score=0.6, microstructure_weight=0.90, # BREAKOUT ) pred_bull = predictor.predict( prices, sentiment=0.85, foreign_flow=0.5, eps_surprise=0.08, signals=sig_bull ) print(f" A. All Bullish:") print(f" Direction: {pred_bull.direction}, Confidence: {pred_bull.confidence}") print(f" Blend scores: {pred_bull.blend_scores}") print(f" Blend weights: {pred_bull.blend_weights}") scenarios.append({ "scenario": "all_bullish", "direction": pred_bull.direction, "confidence": pred_bull.confidence, "blend_scores": pred_bull.blend_scores, }) # Scenario B: All bearish signals sig_bear = SignalInput( ml_score=-0.6, ml_available=False, sentiment_score=-0.7, sentiment_available=True, foreign_flow_score=-0.5, foreign_flow_weight=0.75, earnings_score=-0.4, earnings_weight=1.0, microstructure_score=-0.6, microstructure_weight=0.85, ) pred_bear = predictor.predict( prices, sentiment=0.15, foreign_flow=-0.5, eps_surprise=-0.08, signals=sig_bear ) print(f" B. All Bearish:") print(f" Direction: {pred_bear.direction}, Confidence: {pred_bear.confidence}") scenarios.append({ "scenario": "all_bearish", "direction": pred_bear.direction, "confidence": pred_bear.confidence, }) # Scenario C: Mixed signals (sentiment bullish, flow bearish) sig_mixed = SignalInput( ml_score=0.0, ml_available=False, sentiment_score=0.6, sentiment_available=True, foreign_flow_score=-0.5, foreign_flow_weight=0.50, # WEAK predictivity earnings_score=0.0, earnings_weight=0.35, # STALE microstructure_score=0.3, microstructure_weight=0.60, ) pred_mixed = predictor.predict( prices, sentiment=0.80, foreign_flow=-0.5, eps_surprise=0.0, signals=sig_mixed ) print(f" C. Mixed (sentiment up, flow down):") print(f" Direction: {pred_mixed.direction}, Confidence: {pred_mixed.confidence}") print(f" Dominant signal: {max(pred_mixed.blend_scores.items(), key=lambda x: abs(x[1]))[0]}") scenarios.append({ "scenario": "mixed", "direction": pred_mixed.direction, "confidence": pred_mixed.confidence, "blend_scores": pred_mixed.blend_scores, }) # Scenario D: Only sentiment (other signals absent) sig_sent_only = SignalInput( sentiment_score=0.8, sentiment_available=True, ) pred_sent = predictor.predict( prices, sentiment=0.90, signals=sig_sent_only ) print(f" D. Sentiment Only:") print(f" Direction: {pred_sent.direction}, Confidence: {pred_sent.confidence}") print(f" Active signals: {list(pred_sent.blend_weights.keys())}") scenarios.append({ "scenario": "sentiment_only", "direction": pred_sent.direction, "confidence": pred_sent.confidence, "active_weights": pred_sent.blend_weights, }) return scenarios def main(): print("=" * 60) print("FASE 4 — ENSEMBLE & SUPPORTING SIGNALS EVALUATION") print(f"Timestamp: {datetime.now().isoformat()}") print("=" * 60) all_results = {} # Run all tests all_results["market_microstructure"] = test_market_microstructure() all_results["foreign_flow_predictivity"] = test_foreign_flow_predictivity() all_results["earnings_recency"] = test_earnings_recency() all_results["ensemble_blend"] = test_ensemble_blend() # Summary print("\n" + "=" * 60) print("SUMMARY") print("=" * 60) micro_pass = sum(1 for r in all_results["market_microstructure"] if r["pass"]) earn_pass = sum(1 for r in all_results["earnings_recency"] if r["pass"]) print(f" Market Microstructure: {micro_pass}/5 scenarios passed") print(f" Foreign Flow Predictivity: validated (3 scenarios)") print(f" Earnings Recency: {earn_pass}/4 scenarios passed") print(f" Ensemble Blend: validated (4 scenarios)") # Save results output_dir = project_root / "data" output_dir.mkdir(exist_ok=True) output_path = output_dir / "phase4_evaluation.json" # Add timestamp all_results["metadata"] = { "timestamp": datetime.now().isoformat(), "phase": "FASE 4", "description": "Ensemble & Supporting Signals Enhancement", } with open(output_path, "w") as f: json.dump(all_results, f, indent=2, default=str) print(f"\nResults saved to: {output_path}") if __name__ == "__main__": main()