| self = <test_sector_divergence_recovery.TestRecoveryBounceInDivergence object at 0x1246943e0>
|
| mock_gap = <MagicMock name='compute_multi_day_gap' id='4918154528'>
|
|
|
| @patch(
|
| "rtrader.liquidation.strength_override.compute_multi_day_gap",
|
| return_value=None,
|
| )
|
| def test_divergence_proceeds_when_gap_data_unavailable(self, mock_gap):
|
| """Liquidation should proceed normally when multi-day gap can't be computed."""
|
| detector = SectorDivergenceDetector(
|
| divergence_threshold=0.02,
|
| sell_threshold=0.03,
|
| )
|
|
|
| stock_bars = self._make_intraday_bars(100.0, 104.0, n_bars=60)
|
| sector_bars = self._make_intraday_bars(50.0, 49.75, n_bars=60)
|
|
|
| with patch.object(detector, "_fetch_intraday_data", side_effect=[stock_bars, sector_bars]):
|
| with patch.object(detector, "_detect_thrust_decay", return_value=(False, 0.0, None)):
|
| with patch.object(
|
| detector, "_detect_correlation_breakdown",
|
| return_value=(False, 0.0, None, None, None),
|
| ):
|
| > result = detector.evaluate(
|
| "FAKE",
|
| trading_day=date(2026, 2, 18),
|
| current_time=datetime(2026, 2, 18, 11, 0, tzinfo=None),
|
| sector_etf="XLK",
|
| )
|
|
|
| tests/liquidation/test_sector_divergence_recovery.py:122:
|
| _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
|
|
|
| self = <rtrader.liquidation.sector_divergence.SectorDivergenceDetector object at 0x125253ec0>
|
| symbol = 'FAKE'
|
|
|
| def evaluate(
|
| self,
|
| symbol: str,
|
| *,
|
| trading_day: Optional[date] = None,
|
| current_time: Optional[datetime] = None,
|
| sector_etf: Optional[str] = None,
|
| stock_entry_price: Optional[float] = None,
|
| stock_current_price: Optional[float] = None,
|
| direction: str = "long",
|
| cached_soft_memberships: Optional[List[Any]] = None,
|
| ) -> SectorDivergenceResult:
|
| """Evaluate sector divergence for a stock position.
|
|
|
| Args:
|
| symbol: Stock symbol to evaluate
|
| trading_day: Trading day (default: today)
|
| current_time: Current time for evaluation (default: now)
|
| sector_etf: Sector ETF symbol (auto-detected if not provided)
|
| stock_entry_price: Entry price of the position (for gain calculation)
|
| stock_current_price: Current price (fetched if not provided)
|
| direction: Position direction ("long" or "short")
|
|
|
| Returns:
|
| SectorDivergenceResult with liquidation recommendation
|
|
|
| Direction-aware logic:
|
| - Long: Positive spread (stock > sector) = sell signal (overextended up)
|
| - Short: Negative spread (stock < sector) = cover signal (overextended down)
|
| """
|
| symbol = symbol.upper()
|
|
|
| # Determine sector ETF
|
| if not sector_etf:
|
| sector_etf = get_sector_etf_for_symbol(symbol)
|
| if not sector_etf:
|
| return SectorDivergenceResult(
|
| symbol=symbol,
|
| sector_etf="UNKNOWN",
|
| should_liquidate=False,
|
| divergence_spread=0.0,
|
| confidence=0.0,
|
| reason="Could not determine sector ETF",
|
| )
|
| sector_etf = sector_etf.upper()
|
|
|
| # Determine trading day and time
|
| if trading_day is None:
|
| trading_day = scheduler.get_production_now().date()
|
|
|
| # Market hours
|
| market_open = datetime.combine(trading_day, datetime.min.time()).replace(
|
| hour=9, minute=30, second=0, microsecond=0, tzinfo=NY_TZ
|
| )
|
| market_close = datetime.combine(trading_day, datetime.min.time()).replace(
|
| hour=16, minute=0, second=0, microsecond=0, tzinfo=NY_TZ
|
| )
|
|
|
| if current_time is None:
|
| current_time = scheduler.get_production_now()
|
| if current_time.tzinfo is None:
|
| current_time = current_time.replace(tzinfo=NY_TZ)
|
|
|
| # Clamp to market hours
|
| > if current_time < market_open:
|
| ^^^^^^^^^^^^^^^^^^^^^^^^^^
|
| E TypeError: can't compare offset-naive and offset-aware datetimes
|
|
|
| rtrader/liquidation/sector_divergence.py:245: TypeError
|