diff --git a/src/ccburn/app.py b/src/ccburn/app.py index b5d0f70..8c658f5 100644 --- a/src/ccburn/app.py +++ b/src/ccburn/app.py @@ -49,6 +49,7 @@ def __init__( once: bool = False, compact: bool = False, debug: bool = False, + recent_window_minutes: int | None = None, ): """Initialize the application. @@ -72,6 +73,7 @@ def __init__( self.once = once self.compact = compact self.debug = debug + self.recent_window_minutes = recent_window_minutes # Disable legacy_windows mode for modern terminals to prevent Unicode issues # Rich may incorrectly detect legacy mode even in Windows Terminal @@ -530,6 +532,7 @@ def _run_once(self) -> int: stale_since=stale_since, since_duration=self.since_duration, until=self.until, + recent_window_minutes=self.recent_window_minutes, ) self.console.print(layout) @@ -562,6 +565,7 @@ def _run_tui(self) -> int: error=self.last_error, since_duration=self.since_duration, until=self.until, + recent_window_minutes=self.recent_window_minutes, ) # Set initial window title @@ -626,6 +630,7 @@ def _main_loop(self, live: Live) -> None: stale_since=stale_since, since_duration=self.since_duration, until=self.until, + recent_window_minutes=self.recent_window_minutes, ) live.update(updated_layout) live.refresh() @@ -657,7 +662,7 @@ def _create_json_output(self) -> dict: for lt in [LimitType.SESSION, LimitType.WEEKLY, LimitType.WEEKLY_SONNET]: limit_data = self.last_snapshot.get_limit(lt) if limit_data: - metrics = calculate_burn_metrics(limit_data, self.snapshots) + metrics = calculate_burn_metrics(limit_data, self.snapshots, self.recent_window_minutes) minutes_left = int( (limit_data.resets_at - datetime.now(timezone.utc)).total_seconds() / 60 ) @@ -680,7 +685,7 @@ def _create_json_output(self) -> dict: # Add monthly credits if available monthly_data = self.last_snapshot.monthly if monthly_data: - metrics = calculate_burn_metrics(monthly_data, self.snapshots) + metrics = calculate_burn_metrics(monthly_data, self.snapshots, self.recent_window_minutes) days_left = int( (monthly_data.resets_at - datetime.now(timezone.utc)).total_seconds() / 86400 ) @@ -699,7 +704,7 @@ def _create_json_output(self) -> dict: # Add burn rate and projection for selected limit limit_data = self.last_snapshot.get_limit(self.limit_type) if limit_data: - metrics = calculate_burn_metrics(limit_data, self.snapshots) + metrics = calculate_burn_metrics(limit_data, self.snapshots, self.recent_window_minutes) output["burn_rate"] = { "limit": self.limit_type.value, "percent_per_hour": round(metrics.percent_per_hour, 2), diff --git a/src/ccburn/cli.py b/src/ccburn/cli.py index 8a7e2a0..f2fb893 100644 --- a/src/ccburn/cli.py +++ b/src/ccburn/cli.py @@ -131,6 +131,13 @@ def duration_callback(value: str | None) -> str | None: help="Show debug information including raw API response.", ) +RecentWindowOption = typer.Option( + None, + "--recent-window", + "-r", + help="Limit burn rate calculation to recent data (e.g. '30m', '1h'). Useful after idle periods.", +) + def run_app( limit_type: LimitType | None, @@ -142,6 +149,7 @@ def run_app( interval: int = 5, poll_interval: int = 60, debug: bool = False, + recent_window: str | None = None, ) -> None: """Run the ccburn application with the specified options.""" try: @@ -172,6 +180,14 @@ def run_app( typer.echo(f"Error: {e}", err=True) raise typer.Exit(1) from None + recent_window_minutes = None + if recent_window: + try: + recent_window_minutes = int(parse_duration(recent_window).total_seconds() / 60) + except typer.BadParameter as e: + typer.echo(f"Error: {e}", err=True) + raise typer.Exit(1) from None + app = CCBurnApp( limit_type=limit_type, interval=interval, @@ -182,6 +198,7 @@ def run_app( once=once, compact=compact, debug=debug, + recent_window_minutes=recent_window_minutes, ) exit_code = app.run() @@ -201,6 +218,7 @@ def session( interval: int = SessionIntervalOption, poll_interval: int = PollIntervalOption, debug: bool = DebugOption, + recent_window: str | None = RecentWindowOption, ) -> None: """Display 5-hour rolling session limit. @@ -216,6 +234,7 @@ def session( interval=interval, poll_interval=poll_interval, debug=debug, + recent_window=recent_window, ) @@ -232,6 +251,7 @@ def weekly( interval: int = WeeklyIntervalOption, poll_interval: int = PollIntervalOption, debug: bool = DebugOption, + recent_window: str | None = RecentWindowOption, ) -> None: """Display 7-day weekly limit (all models). @@ -247,6 +267,7 @@ def weekly( interval=interval, poll_interval=poll_interval, debug=debug, + recent_window=recent_window, ) @@ -263,6 +284,7 @@ def weekly_sonnet( interval: int = WeeklyIntervalOption, poll_interval: int = PollIntervalOption, debug: bool = DebugOption, + recent_window: str | None = RecentWindowOption, ) -> None: """Display 7-day weekly limit (Sonnet only). @@ -278,6 +300,7 @@ def weekly_sonnet( interval=interval, poll_interval=poll_interval, debug=debug, + recent_window=recent_window, ) @@ -294,6 +317,7 @@ def monthly( interval: int = MonthlyIntervalOption, poll_interval: int = PollIntervalOption, debug: bool = DebugOption, + recent_window: str | None = RecentWindowOption, ) -> None: """Display monthly credit usage (enterprise accounts). @@ -310,6 +334,7 @@ def monthly( interval=interval, poll_interval=poll_interval, debug=debug, + recent_window=recent_window, ) diff --git a/src/ccburn/display/layout.py b/src/ccburn/display/layout.py index f7df633..2c5668a 100644 --- a/src/ccburn/display/layout.py +++ b/src/ccburn/display/layout.py @@ -76,6 +76,7 @@ def update( stale_since: datetime | None = None, since_duration: timedelta | None = None, until: str = "now", + recent_window_minutes: int | None = None, ) -> Layout: """Update the layout with new data. @@ -98,7 +99,7 @@ def update( # Calculate metrics if limit_data: - self._last_metrics = calculate_burn_metrics(limit_data, snapshots) + self._last_metrics = calculate_burn_metrics(limit_data, snapshots, recent_window_minutes) width, height = self.get_terminal_size() diff --git a/src/ccburn/utils/calculator.py b/src/ccburn/utils/calculator.py index eabca9d..24b6752 100644 --- a/src/ccburn/utils/calculator.py +++ b/src/ccburn/utils/calculator.py @@ -51,6 +51,7 @@ def calculate_burn_rate( window_hours: float, min_points: int = 3, min_span_pct: float = 0.10, + recent_window_minutes: int | None = None, ) -> float: """Calculate burn rate as percentage points per hour using linear regression. @@ -72,8 +73,10 @@ def calculate_burn_rate( if len(snapshots) < 2: return 0.0 - # Use actual window start, not now - window_minutes cutoff = window_start + if recent_window_minutes is not None: + recent_cutoff = datetime.now(timezone.utc) - timedelta(minutes=recent_window_minutes) + cutoff = max(cutoff, recent_cutoff) # Filter to snapshots within window and extract (time, utilization) pairs points: list[tuple[float, float]] = [] # (hours_from_start, utilization_pct) @@ -102,7 +105,8 @@ def calculate_burn_rate( # Cap at 6 hours max so monthly (744h) doesn't require 3+ days of data if first_timestamp and last_timestamp: span_hours = (last_timestamp - first_timestamp).total_seconds() / 3600 - min_span_hours = min(window_hours * min_span_pct, 6.0) + effective_window_hours = (recent_window_minutes / 60) if recent_window_minutes is not None else window_hours + min_span_hours = min(effective_window_hours * min_span_pct, 6.0) if span_hours < min_span_hours: return 0.0 @@ -217,6 +221,7 @@ def get_status(utilization: float, budget_pace: float) -> str: def calculate_burn_metrics( limit_data: LimitData | MonthlyLimitData, snapshots: list[UsageSnapshot], + recent_window_minutes: int | None = None, ) -> BurnMetrics: """Calculate all burn metrics for a limit. @@ -233,6 +238,7 @@ def calculate_burn_metrics( limit_data.limit_type, window_start=limit_data.window_start, window_hours=limit_data.window_hours, + recent_window_minutes=recent_window_minutes, ) time_to_empty = estimate_time_to_empty(limit_data.utilization, burn_rate) trend = classify_burn_trend(burn_rate) diff --git a/tests/test_calculator.py b/tests/test_calculator.py index fed1da6..bce3051 100644 --- a/tests/test_calculator.py +++ b/tests/test_calculator.py @@ -123,6 +123,117 @@ def test_constant_usage(self): assert rate == pytest.approx(0.0, abs=0.1) +class TestRecentWindowBurnRate: + """Tests for calculate_burn_rate with recent_window_minutes.""" + + def _make_snapshot(self, ts: datetime, utilization: float) -> UsageSnapshot: + return UsageSnapshot( + timestamp=ts, + session=LimitData( + utilization=utilization, + resets_at=ts + timedelta(hours=4), + limit_type=LimitType.SESSION, + ), + weekly=None, + weekly_sonnet=None, + weekly_opus=None, + ) + + def test_excludes_old_snapshots(self): + """recent_window_minutes should ignore snapshots older than the cutoff.""" + now = datetime.now(timezone.utc) + window_start = now - timedelta(hours=5) + + # 10 idle snapshots 2h ago (flat at 20%) + old_snapshots = [ + self._make_snapshot(now - timedelta(hours=2, minutes=i), 0.20) + for i in range(10) + ] + # 5 active snapshots in the last 20 minutes (rising from 20% to 30%) + recent_snapshots = [ + self._make_snapshot(now - timedelta(minutes=20 - i * 4), 0.20 + i * 0.02) + for i in range(5) + ] + all_snapshots = sorted(old_snapshots + recent_snapshots, key=lambda s: s.timestamp) + + rate_full = calculate_burn_rate( + all_snapshots, LimitType.SESSION, window_start, window_hours=5 + ) + rate_recent = calculate_burn_rate( + all_snapshots, LimitType.SESSION, window_start, window_hours=5, + recent_window_minutes=30, + ) + + # Recent window sees the active slope; full window is diluted by idle data + assert rate_recent > rate_full + + def test_insufficient_points_returns_zero(self): + """When fewer than min_points snapshots fall in the recent window, return 0.""" + now = datetime.now(timezone.utc) + window_start = now - timedelta(hours=5) + + # 8 snapshots from 2h ago, only 2 in the last 10 minutes + old_snapshots = [ + self._make_snapshot(now - timedelta(hours=2, minutes=i), 0.20 + i * 0.01) + for i in range(8) + ] + recent_snapshots = [ + self._make_snapshot(now - timedelta(minutes=j * 3), 0.50 + j * 0.02) + for j in range(2) + ] + snapshots = sorted(old_snapshots + recent_snapshots, key=lambda s: s.timestamp) + + rate = calculate_burn_rate( + snapshots, LimitType.SESSION, window_start, window_hours=5, + recent_window_minutes=10, + ) + assert rate == 0.0 + + def test_none_is_backward_compatible(self): + """recent_window_minutes=None must give the same result as not passing the param.""" + now = datetime.now(timezone.utc) + window_start = now - timedelta(hours=5) + snapshots = [ + self._make_snapshot(now - timedelta(minutes=30 - i * 6), 0.10 + i * 0.02) + for i in range(5) + ] + + rate_default = calculate_burn_rate( + snapshots, LimitType.SESSION, window_start, window_hours=5 + ) + rate_none = calculate_burn_rate( + snapshots, LimitType.SESSION, window_start, window_hours=5, + recent_window_minutes=None, + ) + assert rate_default == pytest.approx(rate_none) + + def test_span_check_uses_recent_window_size(self): + """min_span check should be relative to recent_window_minutes, not the full window.""" + now = datetime.now(timezone.utc) + window_start = now - timedelta(hours=5) + + # 4 snapshots spanning 4 minutes — well under 10% of a 5h window (30 min) + # but comfortably over 10% of a 60m recent window (6 min... wait, 4 < 6) + # Use a 30m recent window: min_span = 30m * 10% = 3 min, span = 4 min → should pass + snapshots = [ + self._make_snapshot(now - timedelta(minutes=4 - i), 0.30 + i * 0.03) + for i in range(4) + ] + + rate_full = calculate_burn_rate( + snapshots, LimitType.SESSION, window_start, window_hours=5 + ) + rate_recent = calculate_burn_rate( + snapshots, LimitType.SESSION, window_start, window_hours=5, + recent_window_minutes=30, + ) + + # Full window: span (4 min) < min_span (30 min) → 0 + assert rate_full == 0.0 + # Recent window: span (4 min) > min_span (3 min) → non-zero + assert rate_recent > 0.0 + + class TestEstimateTimeToEmpty: """Tests for estimate_time_to_empty."""