From 11022c737d98201c56768e9388506adc52149e08 Mon Sep 17 00:00:00 2001 From: guhyun9454 Date: Wed, 20 May 2026 15:11:56 +0900 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20add=20--recent-window=20option=20to?= =?UTF-8?q?=20limit=20burn=20rate=20calculation=20window?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a --recent-window flag (e.g. 30m, 1h) to all subcommands. When set, burn rate is calculated using only the most recent snapshots, preventing idle periods from diluting the projection line. Defaults to None (full window) for full backward compatibility. --- src/ccburn/app.py | 11 +++- src/ccburn/cli.py | 25 ++++++++ src/ccburn/display/layout.py | 3 +- src/ccburn/utils/calculator.py | 10 ++- tests/test_calculator.py | 111 +++++++++++++++++++++++++++++++++ 5 files changed, 154 insertions(+), 6 deletions(-) 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."""