from __future__ import annotations import argparse import json from collections import Counter from dataclasses import asdict from datetime import datetime, timezone from decimal import Decimal, ROUND_DOWN, InvalidOperation from pathlib import Path from typing import Any from src.execution.cost_model import CostModel from src.execution.order_intent import OrderIntent from src.execution.paper_executor import PaperExecutor from src.risk.pretrade_risk_gate import PretradeRiskGate, RiskContext BRIDGE_ID = "tradingview_paper_bridge" def _json_default(value: Any) -> Any: if isinstance(value, Decimal): return str(value) if isinstance(value, datetime): return value.isoformat() return str(value) def _append_jsonl(path: Path, row: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(row, default=_json_default, sort_keys=True, ensure_ascii=False) + "\n") def _read_jsonl(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] rows: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def _to_decimal(value: Any, default: str = "0") -> Decimal: try: return Decimal(str(value)) except (InvalidOperation, ValueError, TypeError): return Decimal(default) def _quantize_size(notional: Decimal, price: Decimal) -> Decimal: if price <= 0: return Decimal("0") return (notional / price).quantize(Decimal("0.00000001"), rounding=ROUND_DOWN) def _bridge_dir(runtime_dir: Path) -> Path: return runtime_dir / "experiments" / BRIDGE_ID def _processed_ids(path: Path) -> set[str]: return {str(row.get("signal_id")) for row in _read_jsonl(path) if row.get("signal_id")} def _decision_row(signal: dict[str, Any], *, final_decision: str, block_reason: str | None = None, intent: OrderIntent | None = None) -> dict[str, Any]: row: dict[str, Any] = { "timestamp": datetime.now(timezone.utc).isoformat(), "source": "tradingview", "signal_id": signal.get("signal_id"), "strategy_id": signal.get("strategy_id"), "coin": str(signal.get("coin") or "").upper(), "side": signal.get("side"), "action": signal.get("action"), "price": signal.get("price"), "execution_mode": "paper_bridge", "paper_trading": True, "live_order_allowed": False, "mainnet_signed_action": False, "order_intent_created": intent is not None, "final_decision": final_decision, } if block_reason: row["block_reason"] = [block_reason] if intent is not None: row["order_intent"] = asdict(intent) return row def _close_paper_position(signal: dict[str, Any], open_positions: dict[str, Any], bridge_dir: Path) -> tuple[bool, str | None]: coin = str(signal.get("coin") or "").upper() position = open_positions.get(coin) if not isinstance(position, dict): return False, "exit_without_paper_position" exit_price = _to_decimal(signal.get("price")) if exit_price <= 0: return False, "invalid_price" entry_price = _to_decimal(position.get("entry_price")) size = _to_decimal(position.get("size")) entry_fee = _to_decimal(position.get("entry_fee_usd")) side = str(position.get("side") or signal.get("side") or "").lower() if entry_price <= 0 or size <= 0: return False, "invalid_open_position_state" exit_fee = exit_price * size * Decimal("0.00045") gross = (exit_price - entry_price) * size if side == "long" else (entry_price - exit_price) * size net = gross - entry_fee - exit_fee row = { "timestamp": datetime.now(timezone.utc).isoformat(), "event": "paper_exit", "source": "tradingview", "signal_id": signal.get("signal_id"), "entry_signal_id": position.get("signal_id"), "strategy_id": position.get("strategy_id") or signal.get("strategy_id"), "coin": coin, "side": side, "action": "exit", "paper_trading": True, "live_order_allowed": False, "mainnet_signed_action": False, "entry_price": str(entry_price), "exit_price": str(exit_price), "size": str(size), "gross_pnl_usd": str(gross), "entry_fee_usd": str(entry_fee), "exit_fee_usd": str(exit_fee), "net_pnl_usd": str(net), } _append_jsonl(bridge_dir / "trade_journal.jsonl", row) del open_positions[coin] return True, None def _build_entry_intent(signal: dict[str, Any], *, wallet_equity_usdc: Decimal, risk_pct: Decimal, stop_distance_pct: Decimal) -> tuple[OrderIntent | None, str | None]: price = _to_decimal(signal.get("price")) if price <= 0: return None, "invalid_price" side = str(signal.get("side") or "").lower() if side not in {"long", "short"}: return None, "invalid_side" coin = str(signal.get("coin") or "").upper() strategy_id = str(signal.get("strategy_id") or "") if not coin or not strategy_id: return None, "missing_coin_or_strategy" notional = (wallet_equity_usdc * Decimal("0.15")).quantize(Decimal("0.01"), rounding=ROUND_DOWN) risk_usd = (wallet_equity_usdc * risk_pct / Decimal("100")).quantize(Decimal("0.00000001"), rounding=ROUND_DOWN) size = _quantize_size(notional, price) if size <= 0: return None, "invalid_size" order_side = "buy" if side == "long" else "sell" if side == "long": stop_loss = price * (Decimal("1") - stop_distance_pct / Decimal("100")) else: stop_loss = price * (Decimal("1") + stop_distance_pct / Decimal("100")) intent = OrderIntent( strategy_id=strategy_id, symbol=f"{coin}/USDC:USDC", coin=coin, side=order_side, reduce_only=False, order_type="market", tif="Ioc", size=size, price=None, trigger_price=None, stop_loss=stop_loss.quantize(Decimal("0.00000001"), rounding=ROUND_DOWN), take_profit=None, client_order_id=f"tv-paper-{signal.get('signal_id')}", reason=f"tradingview_{side}_entry", risk_usd=risk_usd, estimated_notional_usd=notional, ) return intent, None def process_pending_tradingview_signals( *, runtime_dir: str | Path = "runtime", wallet_equity_usdc: Decimal = Decimal("75"), risk_pct: Decimal = Decimal("0.15"), stop_distance_pct: Decimal = Decimal("1.00"), expected_move_pct: Decimal = Decimal("1.20"), max_parallel_positions: int = 3, ) -> dict[str, Any]: runtime = Path(runtime_dir) bridge_dir = _bridge_dir(runtime) signals = _read_jsonl(runtime / "signals" / "signal_journal.jsonl") processed_path = bridge_dir / "processed_signal_ids.jsonl" processed = _processed_ids(processed_path) state_path = bridge_dir / "state.json" state = json.loads(state_path.read_text(encoding="utf-8")) if state_path.exists() else {"open_positions": {}} open_positions = state.setdefault("open_positions", {}) executor = PaperExecutor(cost_model=CostModel()) gate = PretradeRiskGate(min_order_notional_usd=Decimal("10"), max_parallel_positions=max_parallel_positions, max_order_notional_usd=wallet_equity_usdc * Decimal("0.25"), max_risk_usd=wallet_equity_usdc * Decimal("0.005")) result = {"processed": 0, "paper_trades": 0, "paper_exits": 0, "blocked": {}, "mainnet_signed_action": False, "live_order_allowed": False} blocked: Counter[str] = Counter() for signal in signals: if str(signal.get("source") or "").lower() != "tradingview": continue signal_id = str(signal.get("signal_id") or "") if not signal_id or signal_id in processed: continue result["processed"] += 1 processed.add(signal_id) block_reason: str | None = None intent: OrderIntent | None = None if signal.get("live_order_allowed") is True or signal.get("mainnet_signed_action") is True: block_reason = "unexpected_live_flag" elif signal.get("execution_mode") != "paper_signal": block_reason = "not_paper_signal" elif str(signal.get("action") or "").lower() == "exit": closed, close_block = _close_paper_position(signal, open_positions, bridge_dir) if closed: _append_jsonl(bridge_dir / "signal_decision_journal.jsonl", _decision_row(signal, final_decision="paper_exited")) result["paper_exits"] += 1 else: block_reason = close_block or "exit_failed" elif str(signal.get("action") or "").lower() != "entry": block_reason = "unsupported_action" elif str(signal.get("coin") or "").upper() in open_positions: block_reason = "paper_position_already_open" else: intent, block_reason = _build_entry_intent(signal, wallet_equity_usdc=wallet_equity_usdc, risk_pct=risk_pct, stop_distance_pct=stop_distance_pct) if intent is not None: risk = gate.evaluate(intent, RiskContext(open_positions=len(open_positions), kill_switch_active=False, daily_loss_exceeded=False)) if not risk.allowed: block_reason = ";".join(risk.reasons) or "risk_gate_blocked" intent = None if block_reason: blocked[block_reason] += 1 _append_jsonl(bridge_dir / "signal_decision_journal.jsonl", _decision_row(signal, final_decision=f"blocked:{block_reason}", block_reason=block_reason, intent=intent)) elif intent is not None: mark = _to_decimal(signal.get("price")) fill = executor.execute(intent, mark_price=mark, half_spread_pct=Decimal("0.015"), expected_move_pct=expected_move_pct, depth_penalty_pct=Decimal("0.02"), hold_hours=Decimal("4")) if fill.blocked_by_cost: blocked[fill.blocked_reason] += 1 _append_jsonl(bridge_dir / "signal_decision_journal.jsonl", _decision_row(signal, final_decision=f"blocked:{fill.blocked_reason}", block_reason=fill.blocked_reason, intent=intent)) else: open_positions[intent.coin] = { "signal_id": signal_id, "strategy_id": intent.strategy_id, "side": str(signal.get("side") or ""), "entry_price": str(fill.fill_price), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "entry_fee_usd": str(fill.fee_usd), "timestamp": datetime.now(timezone.utc).isoformat(), } trade_row = { "timestamp": datetime.now(timezone.utc).isoformat(), "event": "paper_entry", "source": "tradingview", "signal_id": signal_id, "strategy_id": intent.strategy_id, "coin": intent.coin, "side": str(signal.get("side") or ""), "action": "entry", "paper_trading": True, "live_order_allowed": False, "mainnet_signed_action": False, "entry_price": str(fill.fill_price), "size": str(intent.size), "stop_loss": str(intent.stop_loss), "estimated_notional_usd": str(intent.estimated_notional_usd), "risk_usd": str(intent.risk_usd), "entry_fee_usd": str(fill.fee_usd), "estimated_roundtrip_cost_pct": str(fill.estimated_roundtrip_cost_pct), "net_pnl_usd": "0", } _append_jsonl(bridge_dir / "trade_journal.jsonl", trade_row) _append_jsonl(bridge_dir / "signal_decision_journal.jsonl", _decision_row(signal, final_decision="paper_entered", intent=intent)) result["paper_trades"] += 1 _append_jsonl(processed_path, {"timestamp": datetime.now(timezone.utc).isoformat(), "signal_id": signal_id, "mainnet_signed_action": False}) state_path.parent.mkdir(parents=True, exist_ok=True) state_path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8") result["blocked"] = dict(blocked) return result def summarize_tradingview_paper_bridge(runtime_dir: str | Path = "runtime") -> dict[str, Any]: bridge_dir = _bridge_dir(Path(runtime_dir)) trades = _read_jsonl(bridge_dir / "trade_journal.jsonl") decisions = _read_jsonl(bridge_dir / "signal_decision_journal.jsonl") entries = [row for row in trades if row.get("event") == "paper_entry"] exits = [row for row in trades if row.get("event") == "paper_exit"] closed_net = sum((_to_decimal(row.get("net_pnl_usd")) for row in exits), Decimal("0")) blockers: Counter[str] = Counter() for row in decisions: final = str(row.get("final_decision") or "") if final.startswith("blocked:"): blockers[final.split(":", 1)[1]] += 1 by_strategy = Counter(str(row.get("strategy_id") or "unknown") for row in entries) by_coin = Counter(str(row.get("coin") or "UNKNOWN").upper() for row in entries) live_count = sum(1 for row in entries + decisions if row.get("live_order_allowed") is True or row.get("mainnet_signed_action") is True) return { "paper_entries": len(entries), "paper_exits": len(exits), "closed_net_pnl_usd": str(closed_net), "blocked": dict(blockers.most_common(5)), "by_strategy": dict(by_strategy.most_common(5)), "by_coin": dict(by_coin.most_common(8)), "live_order_allowed_count": live_count, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Process accepted TradingView signals into paper-only bridge decisions.") parser.add_argument("--runtime-dir", default="runtime") parser.add_argument("--json", action="store_true") args = parser.parse_args(argv) result = process_pending_tradingview_signals(runtime_dir=args.runtime_dir) if args.json: print(json.dumps(result, indent=2, sort_keys=True)) else: print(f"TradingView paper bridge processed={result['processed']} paper_trades={result['paper_trades']} blocked={result['blocked']}") return 0 if __name__ == "__main__": raise SystemExit(main())