#!/usr/bin/env python3 """Realtime IDX Trend Break screener with point-in-time research features. This is a quick screener. Its only operational outputs are ``KANDIDAT FULL PA`` and ``TIDAK LAYAK DITUNGGU``. A candidate still needs 1D/1H/15m alignment and a valid closed 5m trigger before any entry decision. """ import json import os import urllib.request from neobdm_common import datetime, is_market_open from screener_tradingview_common import ( attach_detection_state, build_market_context, build_technical_features, fetch_tradingview_candidates, snapshot_phase, ) from screener_quality import evaluate_quality_features from tradingview_history import load_tradingview_headers from tv_enrichment import atomic_write_json METHOD_NAME = "Patah Tren" SCREENING_FILE = "screening_results.json" OUTPUT_FILE = "trend_break_results.json" DETECTION_STATE_FILE = "trend_break_detection_state.json" MARKET_UNIVERSE_LIMIT = 1000 MAX_RESULTS = 30 MIN_AVG_VALUE = 1_000_000_000 VOL_RATIO_NEAR_MIN = 1.2 VOL_RATIO_CONFIRM_MIN = 1.5 MAX_CHANGE_PCT = 15.0 MIN_CHANGE_PCT = -5.0 MAX_SMA50_STRETCH_PCT = 12.0 DOWNTREND_LOOKBACK = 40 DOWNTREND_BELOW_PCT = 0.60 SWING_LOOKBACK = 20 MIN_LIQUID_SHARES = 500_000 TV_COLUMNS = [ "close", "open", "high", "low", "close[1]", "change", "volume", "average_volume_10d_calc", "relative_volume_10d_calc", "relative_volume_intraday|5", "Value.Traded", "SMA20", "SMA50", "High.1M", "High.3M", "Perf.W", "Perf.1M", "Perf.3M", "RSI", "ADX", "ADX+DI", "ADX-DI", "ATR", "VWAP", "MACD.macd", "MACD.signal", "sector", "industry", ] def _number(value, default=0.0): if isinstance(value, (int, float)): return float(value) return default def detect_trend_break(df): """Historical OHLC detector retained for discovery backtests.""" if df is None or len(df) < 70: return None close = df["Close"] high = df["High"] open_price = df["Open"] volume = df["Volume"] sma20 = close.rolling(20).mean() sma50 = close.rolling(50).mean() avg_vol20 = volume.rolling(20).mean() c = float(close.iloc[-1]) o = float(open_price.iloc[-1]) v = float(volume.iloc[-1]) s20 = float(sma20.iloc[-1]) s50 = float(sma50.iloc[-1]) avg_volume = float(avg_vol20.iloc[-1]) if min(c, o, s20, s50, avg_volume) <= 0: return None prior_close = close.iloc[-(DOWNTREND_LOOKBACK + 1):-1] prior_sma50 = sma50.iloc[-(DOWNTREND_LOOKBACK + 1):-1] valid = prior_sma50.notna() if not valid.any(): return None below_pct = float((prior_close[valid] < prior_sma50[valid]).mean()) if below_pct < DOWNTREND_BELOW_PCT: return None swing_high = float(high.iloc[-(SWING_LOOKBACK + 1):-1].max()) vol_ratio = v / avg_volume if swing_high <= 0 or v < MIN_LIQUID_SHARES or vol_ratio < VOL_RATIO_NEAR_MIN: return None break_margin = (c - swing_high) / swing_high * 100 dist_sma50 = (c - s50) / s50 * 100 confirm = ( c > s20 and c > s50 and c > swing_high and c > o and vol_ratio >= VOL_RATIO_CONFIRM_MIN ) near = ( not confirm and c > s20 and break_margin >= -3.0 and c > o and vol_ratio >= VOL_RATIO_NEAR_MIN ) if not (confirm or near): return None status = "CONFIRM" if confirm else "NEAR" score = ( min(vol_ratio, 5.0) * 15 + min(max(break_margin, -3.0), 10.0) * 4 + below_pct * 20 - (25 if near else 0) ) return { "status": status, "score": round(score, 1), "vol_ratio": round(vol_ratio, 2), "break_margin_pct": round(break_margin, 2), "swing_high": round(swing_high, 2), "dist_sma50_pct": round(dist_sma50, 2), "downtrend_below_pct": round(below_pct * 100, 1), "prior_downtrend": True, "sma20": round(s20, 2), "sma50": round(s50, 2), } def _parse_tradingview_rows(payload): output = {} for item in payload.get("data") or []: values = item.get("d") or [] if len(values) != len(TV_COLUMNS) or not isinstance(values[0], (int, float)): continue row = dict(zip(TV_COLUMNS, values)) ticker = str(item.get("s") or "").split(":")[-1].upper() close = _number(row.get("close")) open_price = _number(row.get("open")) high = _number(row.get("high")) low = _number(row.get("low")) previous_close = _number(row.get("close[1]")) volume = _number(row.get("volume")) avg_volume = _number(row.get("average_volume_10d_calc")) sma20 = _number(row.get("SMA20")) sma50 = _number(row.get("SMA50")) high_1m = _number(row.get("High.1M")) high_3m = _number(row.get("High.3M")) swing_high = high_1m or high_3m if min(close, open_price, sma20, sma50, swing_high) <= 0: continue daily_rvol = ( volume / avg_volume if avg_volume > 0 else _number(row.get("relative_volume_10d_calc")) ) intraday_rvol = _number(row.get("relative_volume_intraday|5")) active_rvol = intraday_rvol if intraday_rvol > 0 else daily_rvol break_margin = (close - swing_high) / swing_high * 100 dist_sma50 = (close - sma50) / sma50 * 100 perf_3m = _number(row.get("Perf.3M")) prior_downtrend = bool(sma20 < sma50 or perf_3m <= -5.0) confirm = ( close > sma20 and close > sma50 and close >= swing_high * 0.995 and close > open_price and active_rvol >= VOL_RATIO_CONFIRM_MIN ) near = ( not confirm and close > sma20 and break_margin >= -3.0 and close > open_price and active_rvol >= VOL_RATIO_NEAR_MIN ) if not (confirm or near): continue status = "CONFIRM" if confirm else "NEAR" score = ( min(active_rvol, 5.0) * 15 + min(max(break_margin, -3.0), 10.0) * 4 + (15 if prior_downtrend else 0) - (25 if near else 0) ) output[ticker] = { "close": close, "price": close, "open": open_price, "high": high, "low": low, "previous_close": previous_close, "change_pct": _number(row.get("change")), "volume": volume, "avg_value": close * avg_volume, "value_traded": _number(row.get("Value.Traded")), "sma20": sma20, "sma50": sma50, "high52": high_3m, "high_1m": high_1m, "perf_week": _number(row.get("Perf.W")), "perf_month": _number(row.get("Perf.1M")), "perf_3m": perf_3m, "rsi": _number(row.get("RSI")), "adx": _number(row.get("ADX")), "plus_di": _number(row.get("ADX+DI")), "minus_di": _number(row.get("ADX-DI")), "atr": _number(row.get("ATR")), "vwap": _number(row.get("VWAP")), "macd": _number(row.get("MACD.macd")), "macd_signal": _number(row.get("MACD.signal")), "sector": str(row.get("sector") or ""), "industry": str(row.get("industry") or ""), "vol_ratio": active_rvol, "vol_ratio_daily": daily_rvol, "vol_ratio_intraday": intraday_rvol, "status": status, "raw_score": round(score, 1), "break_margin_pct": round(break_margin, 3), "swing_high": round(swing_high, 3), "dist_sma50_pct": round(dist_sma50, 3), "prior_downtrend": prior_downtrend, "downtrend_proxy": ( "SMA20_BELOW_SMA50" if sma20 < sma50 else ("PERF_3M_NEGATIVE" if perf_3m <= -5.0 else "MISSING") ), } return output def fetch_tradingview_market_trend_rows( limit=MARKET_UNIVERSE_LIMIT, with_total=False, ): payload = { "filter": [ {"left": "market_cap_basic", "operation": "nempty"}, {"left": "type", "operation": "in_range", "right": ["stock", "dr", "fund"]}, ], "options": {"lang": "en"}, "markets": ["indonesia"], "symbols": {"query": {"types": []}, "tickers": []}, "columns": TV_COLUMNS, "sort": {"sortBy": "Value.Traded", "sortOrder": "desc"}, "range": [0, limit], } request = urllib.request.Request( "https://scanner.tradingview.com/indonesia/scan", data=json.dumps(payload).encode("utf-8"), headers=load_tradingview_headers(), ) with urllib.request.urlopen(request, timeout=25) as response: raw = json.loads(response.read().decode("utf-8")) rows = _parse_tradingview_rows(raw) if with_total: return rows, int(raw.get("totalCount") or len(raw.get("data") or [])) return rows def fetch_tradingview_trend_rows(tickers, chunk_size=100): """Explicit ticker path retained for tests and focused diagnostics.""" output = {} headers = load_tradingview_headers() for start in range(0, len(tickers), chunk_size): chunk = tickers[start:start + chunk_size] request = urllib.request.Request( "https://scanner.tradingview.com/indonesia/scan", data=json.dumps({ "symbols": { "tickers": [f"IDX:{ticker}" for ticker in chunk], "query": {"types": []}, }, "columns": TV_COLUMNS, }).encode("utf-8"), headers=headers, ) with urllib.request.urlopen(request, timeout=20) as response: output.update(_parse_tradingview_rows( json.loads(response.read().decode("utf-8")) )) return output def selection_quality(row): reasons = [] warnings = [] change = _number(row.get("change_pct")) avg_value = _number(row.get("avg_value")) dist_sma50 = _number(row.get("dist_sma50_pct")) rsi = _number(row.get("rsi")) if not row.get("prior_downtrend"): reasons.append("Tidak ada bukti downtrend sebelumnya") if avg_value < MIN_AVG_VALUE: reasons.append( f"Likuiditas rata-rata Rp {avg_value / 1_000_000:.0f} juta " f"(minimum Rp {MIN_AVG_VALUE / 1_000_000_000:.0f} miliar)" ) if change <= MIN_CHANGE_PCT: reasons.append(f"Dump harian {change:+.1f}%") if change >= MAX_CHANGE_PCT: reasons.append(f"Sudah naik {change:+.1f}%") if dist_sma50 > MAX_SMA50_STRETCH_PCT: reasons.append(f"Extended {dist_sma50:+.1f}% dari SMA50") if rsi > 72: reasons.append(f"RSI daily terlalu tinggi {rsi:.1f}") elif rsi > 67: warnings.append(f"RSI mulai tinggi {rsi:.1f}") if row.get("status") == "NEAR": warnings.append("Belum close breakout; hanya kandidat pemantauan") penalty = max(0.0, dist_sma50 - 4.0) * 1.5 penalty += max(0.0, change - 10.0) * 2.0 selection_score = max(0.0, _number(row.get("raw_score")) - penalty) return { "screening_status": ( "TIDAK LAYAK DITUNGGU" if reasons else "KANDIDAT FULL PA" ), "screening_reasons": reasons or warnings or [ "Lolos downtrend, breakout proximity, volume, dan liquidity gate" ], "selection_score": round(selection_score, 1), } def trend_technical_features(row, market_context): features = build_technical_features(row, market_context) features.update({ "signal_class": row.get("status"), "prior_downtrend": bool(row.get("prior_downtrend")), "downtrend_proxy": row.get("downtrend_proxy") or "MISSING", "perf_3m_pct": round(_number(row.get("perf_3m")), 3), "break_margin_pct": round(_number(row.get("break_margin_pct")), 3), "dist_sma50_pct": round(_number(row.get("dist_sma50_pct")), 3), "swing_high": round(_number(row.get("swing_high")), 3), }) return features def main(): print(f"๐Ÿ”Ž Scan {METHOD_NAME}: universe pasar IDX via TradingView") market_rows, universe_count = fetch_tradingview_market_trend_rows( with_total=True ) if not market_rows: raise RuntimeError( "TradingView mengembalikan universe kosong; hasil lama dipertahankan." ) print( f"๐Ÿ“ก Universe terbaca: {universe_count} ticker; " f"{len(market_rows)} mendekati/mengonfirmasi break" ) try: composite = fetch_tradingview_candidates(["COMPOSITE"]).get("COMPOSITE") or {} except Exception as exc: print(f"โš ๏ธ Konteks IHSG gagal: {exc}") composite = {} market_context = build_market_context(composite) now = datetime.now() phase = snapshot_phase(now) qualified = [] rejected = [] screening_rows = [] for ticker, row in market_rows.items(): quality = selection_quality(row) technical_features = trend_technical_features(row, market_context) detail = { "ticker": ticker, "price": round(row["price"], 3), "change_pct": round(row["change_pct"], 3), "status": row["status"], "score": row["raw_score"], "vol_ratio": round(row["vol_ratio"], 3), "vol_ratio_daily": round(row["vol_ratio_daily"], 3), "vol_ratio_intraday": round(row["vol_ratio_intraday"], 3), "break_margin_pct": row["break_margin_pct"], "swing_high": row["swing_high"], "dist_sma50_pct": row["dist_sma50_pct"], "downtrend_below_pct": 100.0 if row["prior_downtrend"] else 0.0, "prior_downtrend": row["prior_downtrend"], "downtrend_proxy": row["downtrend_proxy"], "sma20": round(row["sma20"], 3), "sma50": round(row["sma50"], 3), "avg_value": round(row["avg_value"], 2), "technical_features": technical_features, # Mode Kualitas 85+ adalah filter riset yang disimpan per snapshot; # bukan gate screening dan bukan rekomendasi entry. "quality_protocol": evaluate_quality_features( technical_features, signal={ "change_pct": row["change_pct"], "vol_ratio": row["vol_ratio"], "vol_ratio_intraday": row["vol_ratio_intraday"], "value_traded": row["avg_value"], }, ), **quality, } if quality["screening_status"] == "KANDIDAT FULL PA": qualified.append(detail) else: rejected.append(detail) screening_rows.append({ "symbol": ticker, "price": str(int(row["price"])), "dn-0": str(round(row["change_pct"], 2)), ">ma5": "-", ">ma10": "-", ">ma20": "v", "unusual": "v" if row["vol_ratio"] >= VOL_RATIO_CONFIRM_MIN else "-", "likuid": "v" if row["avg_value"] >= MIN_AVG_VALUE else "-", "source_note": ( f"Patah tren {row['status']}: swing high {row['swing_high']} " f"({row['break_margin_pct']:+.2f}%), RVOL aktif {row['vol_ratio']:.2f}x, " f"prior {row['downtrend_proxy']}" ), }) qualified.sort( key=lambda item: (item["selection_score"], item["score"], item["ticker"]), reverse=True, ) rejected.sort( key=lambda item: (item["selection_score"], item["score"], item["ticker"]), reverse=True, ) for rank, detail in enumerate(qualified, start=1): detail["selection_rank"] = rank qualified, entered, exited, detection_state = attach_detection_state( qualified, now=now, state_path=DETECTION_STATE_FILE, output_path=OUTPUT_FILE, phase=phase, market_rows=market_rows, ) # Setelah snapshot LIVE pertama, cohort hanya bertambah sampai pergantian # tanggal. Jangan memotongnya lagi agar saham yang sudah terdeteksi tidak # keluar pada hari yang sama. sticky_live_cohort = bool(detection_state.get("has_live_snapshot")) display = qualified if sticky_live_cohort else qualified[:MAX_RESULTS] shown = {item["ticker"] for item in display} screening_rows = [ row for row in screening_rows if row["symbol"] in shown ] score_by_ticker = { item["ticker"]: item["selection_score"] for item in display } screening_rows.sort( key=lambda row: score_by_ticker.get(row["symbol"], 0), reverse=True, ) confirm_lock_tickers = sorted( (detection_state.get("confirm_locks") or {}).keys() ) try: with open(SCREENING_FILE, encoding="utf-8") as handle: screening = json.load(handle) except (FileNotFoundError, json.JSONDecodeError, OSError): screening = {"results_by_method": {}} screening.setdefault("results_by_method", {})[METHOD_NAME] = screening_rows screening.setdefault("method_timestamps", {})[METHOD_NAME] = now.isoformat( timespec="seconds" ) screening["timestamp"] = now.strftime("%Y-%m-%d %H:%M:%S") atomic_write_json(SCREENING_FILE, screening, indent=2) payload = { "timestamp": screening["timestamp"], "timestamp_iso": now.isoformat(timespec="seconds"), "source": "TradingView", "feature_schema_version": "2.0", "scan_mode": "market-wide-auto-capable", "market_open": is_market_open(now), "snapshot_phase": phase, "market_context": market_context, "universe_count": universe_count, "candidate_count": len(qualified), "display_count": len(display), "rejected_count": len(rejected), "entered_tickers": entered, "exited_tickers": exited, "confirm_lock": { "scope": "CURRENT_SESSION", "tickers": confirm_lock_tickers, "new_tickers": detection_state.get("confirm_entered_tickers") or [], "count": len(confirm_lock_tickers), "note": ( "Ticker tetap tampil setelah pertama kali berstatus CONFIRM " "selama sesi berjalan; quote dan metrik terbaru tetap dapat berubah." ), }, "signals": display, "all_qualified_signals": qualified, "rejected_signals": rejected, "selection_policy": { "display_limit": MAX_RESULTS, "min_avg_value": MIN_AVG_VALUE, "min_change_pct_exclusive": MIN_CHANGE_PCT, "max_change_pct_exclusive": MAX_CHANGE_PCT, "max_sma50_stretch_pct": MAX_SMA50_STRETCH_PCT, "screening_outputs": [ "KANDIDAT FULL PA", "TIDAK LAYAK DITUNGGU", ], "note": ( "CONFIRM/NEAR adalah kelas pola daily, bukan status entry. " "SIAP ENTRY tetap memerlukan PA 1D/1H/15m dan trigger close 5m." ), }, } from trend_break_history import archive_daily_snapshot archive, archive_path = archive_daily_snapshot(payload) payload["history"] = { "date": archive["date"], "file": os.path.relpath( archive_path, os.path.dirname(os.path.abspath(__file__)), ), "snapshot_count": archive["snapshot_count"], "ticker_count": archive["ticker_count"], } atomic_write_json(OUTPUT_FILE, payload, indent=2) try: from trend_break_pa_alert import notify_trend_break_pa_candidates alert_result = notify_trend_break_pa_candidates(payload) print( "๐Ÿ“ฃ PA Telegram Trend Break: " f"status={alert_result.get('status')} " f"candidate={alert_result.get('candidates', 0)} " f"terkirim={alert_result.get('sent', 0)} " f"dedupe={alert_result.get('skipped', 0)}" ) except Exception as exc: # Alert failure must not discard the market snapshot. print(f"โš ๏ธ PA Telegram Trend Break gagal: {type(exc).__name__}") print( f"๐Ÿ—‚๏ธ Arsip {payload['history']['file']}: " f"{archive['snapshot_count']} snapshot, {archive['ticker_count']} ticker" ) try: from trend_break_outcomes import refresh_outcomes outcomes, refreshed = refresh_outcomes() summary = outcomes.get("summary") or {} print( f"๐ŸŽฏ Outcome {'diperbarui' if refreshed else 'masih fresh'}: " f"D+1={summary.get('d1_labeled', 0)}, " f"D+3={summary.get('d3_labeled', 0)}, " f"pending={summary.get('pending_d1', 0)}" ) except Exception as exc: print(f"โš ๏ธ Outcome belum diperbarui: {exc}") try: from trend_break_audit import write_audit audit, _ = write_audit() dataset = audit.get("dataset") or {} print( f"๐Ÿงพ Audit {audit.get('status')}: " f"LIVE berlabel={dataset.get('forward_d1_labeled_records', 0)}, " f"pending={dataset.get('forward_live_pending_records', 0)}" ) except Exception as exc: print(f"โš ๏ธ Audit belum diperbarui: {exc}") if os.environ.get("TRENDBREAK_RUN_BACKTEST", "0").lower() in { "1", "true", "yes", "on", }: try: from backtest_trend_break import ensure_backtest backtest = ensure_backtest() print( f"๐Ÿ“Š Backtest discovery: WR {backtest.get('win_rate')}% " f"N={backtest.get('closed_trades', 0)}" ) except Exception as exc: print(f"โš ๏ธ Backtest discovery gagal: {exc}") print( f"โœ… {len(qualified)} kandidat lolos, {len(rejected)} ditolak, " f"Top {len(display)} ditampilkan; fase={phase}" ) for item in display: print( f" {item['ticker']}: {item['status']} " f"selection={item['selection_score']} " f"break={item['break_margin_pct']:+.2f}% " f"rvol={item['vol_ratio']:.2f}x" ) if __name__ == "__main__": main()