✨ add --recent-window option to limit burn rate calculation window

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),

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,
         )
 
 

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()
 

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)

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."""