Merge pull request #26 from Frostbite1536/claude/expert-bug-audit-prompt-CmI3w

Claude/expert bug audit prompt cm i3w

Author: Frostbite1536

EXPERT_BUG_AUDIT_FINDINGS.md

@@ -9,7 +9,7 @@
 from .trade_loader import Trade
 
 
-def _normalize_datetime(dt) -> datetime:
+def _normalize_datetime(dt) -> Optional[datetime]:
     """
     Normalize a datetime value to a naive datetime for consistent comparison.
     Handles both timezone-aware and naive datetimes, and numeric timestamps.

EXPERT_BUG_AUDIT_PROMPT.md

@@ -6,7 +6,7 @@
 from decimal import Decimal
 from typing import List, Dict
 import pandas as pd
-from .trade_loader import Trade
+from .trade_loader import Trade, sanitize_numeric
 from .inference import detect_market_resolution
 
 
@@ -38,16 +38,17 @@ def calculate_pnl(trades: List[Trade]) -> pd.DataFrame:
         cumulative.append(float(running))
     df["cumulative_pnl"] = cumulative
 
-    # Calculate exposure (net shares held)
-    df["exposure"] = 0.0
-    cumulative_shares = 0.0
+    # Calculate exposure (net shares held) using Decimal to avoid float drift
+    exposure_values = []
+    cumulative_shares = Decimal("0")
 
     for idx, row in df.iterrows():
         if row["type"] in ["Buy", "Market Buy", "Limit Buy"]:
-            cumulative_shares += row["shares"]
+            cumulative_shares += Decimal(str(row["shares"]))
         elif row["type"] in ["Sell", "Market Sell", "Limit Sell"]:
-            cumulative_shares -= row["shares"]
-        df.at[idx, "exposure"] = cumulative_shares
+            cumulative_shares -= Decimal(str(row["shares"]))
+        exposure_values.append(float(cumulative_shares))
+    df["exposure"] = exposure_values
 
     return df
 
@@ -170,10 +171,12 @@ def calculate_global_pnl_summary(trades: List[Trade]) -> Dict:
     if len(sources) > 1:
         for source in sources:
             source_trades = [t for t in trades if getattr(t, "source", "limitless") == source]
-            source_pnl = sum(t.pnl for t in source_trades)
+            source_pnl = Decimal("0")
+            for t in source_trades:
+                source_pnl += Decimal(str(t.pnl))
             by_source[source] = {
                 "total_trades": len(source_trades),
-                "total_pnl": source_pnl,
+                "total_pnl": float(source_pnl),
                 "currency": (
                     getattr(source_trades[0], "currency", "USD") if source_trades else "USD"
                 ),
@@ -190,7 +193,10 @@ def calculate_market_pnl(trades: List[Trade]) -> Dict[str, Dict]:
     Returns:
         Dictionary mapping market_slug to PnL statistics
     """
-    market_stats = {}
+    market_stats: Dict[str, Dict] = {}
+    # Use Decimal accumulators to avoid float drift, keyed by slug
+    _volume_acc: Dict[str, Decimal] = {}
+    _pnl_acc: Dict[str, Decimal] = {}
 
     for trade in trades:
         slug = trade.market_slug
@@ -201,11 +207,17 @@ def calculate_market_pnl(trades: List[Trade]) -> Dict[str, Dict]:
                 "total_pnl": 0.0,
                 "trade_count": 0,
             }
+            _volume_acc[slug] = Decimal("0")
+            _pnl_acc[slug] = Decimal("0")
 
-        market_stats[slug]["total_volume"] += trade.cost
-        market_stats[slug]["total_pnl"] += trade.pnl
+        _volume_acc[slug] += Decimal(str(trade.cost))
+        _pnl_acc[slug] += Decimal(str(trade.pnl))
         market_stats[slug]["trade_count"] += 1
 
+    for slug in market_stats:
+        market_stats[slug]["total_volume"] = sanitize_numeric(float(_volume_acc[slug]))
+        market_stats[slug]["total_pnl"] = sanitize_numeric(float(_pnl_acc[slug]))
+
     return market_stats
 
 

prediction_analyzer/filters.py

@@ -19,7 +19,7 @@
 from typing import List, Optional, Dict, Any
 
 from .base import MarketProvider
-from ..trade_loader import Trade, _parse_timestamp
+from ..trade_loader import Trade, _parse_timestamp, sanitize_numeric
 
 logger = logging.getLogger(__name__)
 
@@ -176,14 +176,16 @@ def _fetch_position_pnl(self, api_key: str) -> Dict[str, float]:
 
             for pos in data.get("market_positions", []):
                 ticker = pos.get("ticker", "")
-                pnl_map[ticker] = float(pos.get("realized_pnl_dollars", "0"))
+                pnl_map[ticker] = sanitize_numeric(float(pos.get("realized_pnl_dollars", "0")))
 
             cursor = data.get("cursor", "")
             if not cursor:
                 break
 
         return pnl_map
 
+    _SELL_TYPES = {"sell", "market sell", "limit sell"}
+
     @staticmethod
     def _apply_position_pnl(trades: List[Trade], pnl_map: Dict[str, float]):
         """Distribute realized PnL from positions across sell trades,
@@ -192,11 +194,11 @@ def _apply_position_pnl(trades: List[Trade], pnl_map: Dict[str, float]):
 
         sell_shares: Dict[str, float] = defaultdict(float)
         for t in trades:
-            if t.type.lower() == "sell" and t.market_slug in pnl_map:
+            if t.type.lower() in KalshiProvider._SELL_TYPES and t.market_slug in pnl_map:
                 sell_shares[t.market_slug] += t.shares
 
         for t in trades:
-            if t.type.lower() == "sell" and t.market_slug in pnl_map:
+            if t.type.lower() in KalshiProvider._SELL_TYPES and t.market_slug in pnl_map:
                 total = sell_shares[t.market_slug]
                 if total > 0:
                     t.pnl = pnl_map[t.market_slug] * (t.shares / total)
@@ -266,7 +268,7 @@ def normalize_trade(self, raw: dict, **kwargs) -> Trade:
                 )
                 price = 0.0
 
-        count_str = raw.get("count_fp", str(raw.get("count", 0)))
+        count_str = raw.get("count_fp") or str(raw.get("count") or 0)
         try:
             count = float(count_str)
         except (ValueError, TypeError):

prediction_analyzer/pnl.py

@@ -115,9 +115,14 @@ def calculate_capital_gains(
                 # Only record transaction details for sells in the tax year
                 if in_tax_year:
                     matched_d = Decimal(str(matched_shares))
-                    cost_basis = float(matched_d * lot["cost_per_share"])
-                    proceeds = float(matched_d * proceeds_per_share)
-                    gain_loss = proceeds - cost_basis
+                    # Keep in Decimal for precise subtraction, convert to float at the end
+                    cost_basis_d = matched_d * lot["cost_per_share"]
+                    proceeds_d = matched_d * proceeds_per_share
+                    gain_loss_d = proceeds_d - cost_basis_d
+
+                    cost_basis = float(cost_basis_d)
+                    proceeds = float(proceeds_d)
+                    gain_loss = float(gain_loss_d)
 
                     # Determine holding period
                     holding_delta = trade.timestamp - lot["date"]
@@ -138,8 +143,6 @@ def calculate_capital_gains(
                     if sell_fee > 0:
                         tx["fee"] = sanitize_numeric(float(sell_fee))
                     transactions.append(tx)
-
-                    gain_loss_d = Decimal(str(gain_loss))
                     if is_long_term:
                         if gain_loss_d >= 0:
                             long_term_gains += gain_loss_d

prediction_analyzer/providers/kalshi.py

@@ -25,6 +25,7 @@
 from ..errors import safe_tool
 from ..serializers import to_json_text
 from ..validators import validate_chart_type, validate_market_slug
+from prediction_analyzer.exceptions import ChartError
 
 logger = logging.getLogger(__name__)
 
@@ -112,6 +113,12 @@ async def _handle_generate_chart(arguments: dict):
         raise ValueError("chart_type is required")
 
     validate_chart_type(chart_type)
+    if chart_type not in _CHART_GENERATORS:
+        raise ChartError(
+            f"Chart type '{chart_type}' is not supported for single-market charts. "
+            f"Use 'generate_dashboard' for global charts, or choose from: "
+            f"{sorted(_CHART_GENERATORS.keys())}"
+        )
     validate_market_slug(market_slug, get_unique_markets(session.trades))
 
     trades = filter_trades_by_market_slug(session.trades, market_slug)

prediction_analyzer/tax.py

@@ -87,14 +87,21 @@ def validate_export_format(fmt: str) -> str:
 
 
 def validate_positive_int(value: Optional[int], param_name: str) -> Optional[int]:
-    """Validate that an integer parameter is positive."""
+    """Validate that an integer parameter is positive.
+
+    Accepts float-typed integers (e.g. 5.0) since JSON doesn't
+    distinguish between 5 and 5.0, and LLMs commonly send these.
+    """
     if value is None:
         return None
     if isinstance(value, float):
         if math.isnan(value) or math.isinf(value):
             raise InvalidFilterError(
                 f"Invalid {param_name}: {value}. Must be a positive integer, not NaN/Infinity."
             )
+        # Accept integral floats like 5.0 by converting to int
+        if value == int(value):
+            value = int(value)
     if not isinstance(value, int) or value < 1:
         raise InvalidFilterError(f"Invalid {param_name}: {value}. Must be a positive integer.")
     return value