diff --git a/CLAUDE.md b/CLAUDE.md index 4518e07..3ed4dd9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -159,18 +159,20 @@ The codebase follows a four-layer architecture: ## Key Files - `nadlan_mcp/govmap/` - **✅ Refactored modular package** (Phase 3 & 4 complete) - - `models.py` - **✨ Pydantic v2 data models** (~338 lines, 9 models) **NEW in v2.0** + - `models.py` - **✨ Pydantic v2 data models** (~400 lines, 10 models) **UPDATED** (added OutlierReport) - `client.py` - Core API client (~30KB, returns Pydantic models) - `validators.py` - Input validation (~3KB) - `filters.py` - Deal filtering (~5KB, accepts/returns models) - - `statistics.py` - Statistical calculations (~4KB, returns models) - - `market_analysis.py` - Market analysis (~17KB, returns models) + - `statistics.py` - Statistical calculations with outlier filtering (~8KB, returns models) **UPDATED** + - `market_analysis.py` - Market analysis with robust volatility (~18KB, returns models) **UPDATED** + - `outlier_detection.py` - **✨ Outlier detection & filtering** (~10KB) **NEW** - `utils.py` - Helper utilities (~4KB) - `__init__.py` - Package exports for backward compatibility - `nadlan_mcp/fastmcp_server.py` - MCP tool definitions (10 implemented tools) - `nadlan_mcp/config.py` - Configuration management - `run_fastmcp_server.py` - Server entry point - `tests/govmap/test_models.py` - **✨ Model tests** (50+ tests) **NEW** +- `tests/govmap/test_outlier_detection.py` - **✨ Outlier detection tests** (24 tests) **NEW** - `tests/test_govmap_client.py` - Main test suite (34 tests, partially updated for v2.0) - `MIGRATION.md` - **✨ v1.x → v2.0 migration guide** **NEW** - `USECASES.md` - **Product roadmap and feature status** (essential reading) @@ -233,6 +235,69 @@ deal_json = deal.model_dump_json() # Convert to JSON string - Use `.model_dump()` to serialize models to dicts for JSON/MCP responses - See `MIGRATION.md` for complete migration guide +### Outlier Detection & Statistical Refinement **✨ NEW** + +The MCP now includes configurable outlier detection and robust statistical measures to improve analysis accuracy. This is especially important for handling: +- **Data entry errors**: Wrong area values (1 sqm instead of 100) → extreme price_per_sqm +- **Partial deals**: Family sales or partial ownership (400K apartment when typical is 1.6M) +- **Special circumstances**: Distressed sales, data anomalies + +**Key Features:** +- **IQR-based outlier detection**: Uses Interquartile Range (robust to skewed data) +- **Hard bounds filtering**: Removes obvious errors (price_per_sqm < 1K or > 100K NIS/sqm, deals < 100K NIS) +- **Robust volatility**: Investment analysis uses IQR instead of std_dev for stability ratings +- **Transparent reporting**: Returns both filtered and unfiltered statistics with outlier reports + +**Configuration (environment variables):** +```bash +# Outlier Detection Strategy +ANALYSIS_OUTLIER_METHOD=iqr # Options: iqr, percent, none (default: iqr) +ANALYSIS_IQR_MULTIPLIER=1.5 # 1.5=moderate, 3.0=conservative (default: 1.5) +ANALYSIS_MIN_DEALS_FOR_OUTLIER_DETECTION=10 # Minimum deals needed (default: 10) + +# Hard Bounds (catches obvious data errors) +ANALYSIS_PRICE_PER_SQM_MIN=1000 # 1K NIS/sqm minimum (default: 1000) +ANALYSIS_PRICE_PER_SQM_MAX=100000 # 100K NIS/sqm maximum (default: 100000) +ANALYSIS_MIN_DEAL_AMOUNT=100000 # 100K NIS minimum (catches partial deals, default: 100000) + +# Statistical Robustness (for investment analysis) +ANALYSIS_USE_ROBUST_VOLATILITY=true # Use IQR for volatility (default: true) +ANALYSIS_USE_ROBUST_TRENDS=true # Filter outliers before trend analysis (default: true) + +# Reporting +ANALYSIS_INCLUDE_UNFILTERED_STATS=true # Report both filtered/unfiltered (default: true) +``` + +**Usage Example:** +```python +from nadlan_mcp.govmap.statistics import calculate_deal_statistics +from nadlan_mcp.config import GovmapConfig + +# With outlier filtering (default behavior) +config = GovmapConfig() # Uses env vars or defaults +stats = calculate_deal_statistics(deals, config) + +# Access filtered statistics (main results after outlier removal) +filtered_mean = stats.filtered_price_per_sqm_statistics["mean"] +filtered_median = stats.filtered_price_per_sqm_statistics["median"] + +# Access outlier report +if stats.outlier_report: + print(f"Removed {stats.outlier_report.outliers_removed} outliers") + print(f"Method: {stats.outlier_report.method_used}") + print(f"Filtered indices: {stats.outlier_report.outlier_indices}") + +# Original (unfiltered) statistics still available +original_mean = stats.price_per_sqm_statistics["mean"] +``` + +**Design Principles:** +- **Enabled by default**: Moderate filtering (k=1.5) to catch common errors +- **Transparent**: Both filtered and unfiltered statistics returned +- **Conservative with real data**: Hard bounds + IQR preserve legitimate high-end properties +- **Configurable**: Adjust sensitivity via environment variables +- **MCP provides data, LLM provides intelligence**: Outlier detection improves data quality; LLM interprets results + ### Retry Logic All API calls use automatic retry with exponential backoff (configurable via `GOVMAP_MAX_RETRIES`). The pattern is implemented in `GovmapClient._make_request()`. diff --git a/nadlan_mcp/config.py b/nadlan_mcp/config.py index 911fc0a..1b7f666 100644 --- a/nadlan_mcp/config.py +++ b/nadlan_mcp/config.py @@ -56,6 +56,45 @@ class GovmapConfig: default_factory=lambda: int(os.getenv("GOVMAP_MAX_POLYGONS", "10")) ) + # Outlier Detection & Statistical Refinement + analysis_outlier_method: str = field( + default_factory=lambda: os.getenv("ANALYSIS_OUTLIER_METHOD", "iqr") + ) + analysis_iqr_multiplier: float = field( + default_factory=lambda: float(os.getenv("ANALYSIS_IQR_MULTIPLIER", "1.5")) + ) + analysis_min_deals_for_outlier_detection: int = field( + default_factory=lambda: int(os.getenv("ANALYSIS_MIN_DEALS_FOR_OUTLIER_DETECTION", "10")) + ) + + # Hard Bounds for Price per Sqm (catches obvious data errors) + analysis_price_per_sqm_min: float = field( + default_factory=lambda: float(os.getenv("ANALYSIS_PRICE_PER_SQM_MIN", "1000")) + ) + analysis_price_per_sqm_max: float = field( + default_factory=lambda: float(os.getenv("ANALYSIS_PRICE_PER_SQM_MAX", "100000")) + ) + + # Hard Bounds for Deal Amount (catches partial deals) + analysis_min_deal_amount: float = field( + default_factory=lambda: float(os.getenv("ANALYSIS_MIN_DEAL_AMOUNT", "100000")) + ) + + # Statistical Robustness (for investment analysis) + analysis_use_robust_volatility: bool = field( + default_factory=lambda: os.getenv("ANALYSIS_USE_ROBUST_VOLATILITY", "true").lower() + == "true" + ) + analysis_use_robust_trends: bool = field( + default_factory=lambda: os.getenv("ANALYSIS_USE_ROBUST_TRENDS", "true").lower() == "true" + ) + + # Reporting + analysis_include_unfiltered_stats: bool = field( + default_factory=lambda: os.getenv("ANALYSIS_INCLUDE_UNFILTERED_STATS", "true").lower() + == "true" + ) + # User agent user_agent: str = field( default_factory=lambda: os.getenv("GOVMAP_USER_AGENT", "NadlanMCP/1.0.0") @@ -92,6 +131,20 @@ class GovmapConfig: if not self.user_agent: raise ValueError("user_agent cannot be empty") + # Validate outlier detection settings + if self.analysis_outlier_method not in ["iqr", "percent", "none"]: + raise ValueError("analysis_outlier_method must be one of: iqr, percent, none") + if self.analysis_iqr_multiplier <= 0: + raise ValueError("analysis_iqr_multiplier must be positive") + if self.analysis_min_deals_for_outlier_detection < 0: + raise ValueError("analysis_min_deals_for_outlier_detection must be non-negative") + if self.analysis_price_per_sqm_min <= 0: + raise ValueError("analysis_price_per_sqm_min must be positive") + if self.analysis_price_per_sqm_max <= self.analysis_price_per_sqm_min: + raise ValueError("analysis_price_per_sqm_max must be > analysis_price_per_sqm_min") + if self.analysis_min_deal_amount <= 0: + raise ValueError("analysis_min_deal_amount must be positive") + # Global configuration instance _config: Optional[GovmapConfig] = None diff --git a/nadlan_mcp/govmap/market_analysis.py b/nadlan_mcp/govmap/market_analysis.py index ecd0ef5..711f0a9 100644 --- a/nadlan_mcp/govmap/market_analysis.py +++ b/nadlan_mcp/govmap/market_analysis.py @@ -10,7 +10,10 @@ from datetime import date, datetime, timedelta import logging from typing import Dict, List, Optional, Tuple +from nadlan_mcp.config import GovmapConfig, get_config + from .models import Deal, InvestmentAnalysis, LiquidityMetrics, MarketActivityScore +from .outlier_detection import calculate_iqr, filter_deals_for_analysis from .statistics import calculate_std_dev logger = logging.getLogger(__name__) @@ -201,7 +204,9 @@ def calculate_market_activity_score( ) -def analyze_investment_potential(deals: List[Deal]) -> InvestmentAnalysis: +def analyze_investment_potential( + deals: List[Deal], config: Optional[GovmapConfig] = None +) -> InvestmentAnalysis: """ Analyze investment potential based on price trends and market stability. @@ -209,8 +214,12 @@ def analyze_investment_potential(deals: List[Deal]) -> InvestmentAnalysis: and provides investment metrics for decision-making. The MCP provides data metrics; the LLM interprets them for investment advice. + With outlier filtering enabled, this function removes statistical outliers + before calculating volatility and trends, resulting in more accurate assessments. + Args: deals: List of Deal model instances with price and date information + config: Configuration object (optional, uses global config if not provided) Returns: InvestmentAnalysis model containing: @@ -229,9 +238,27 @@ def analyze_investment_potential(deals: List[Deal]) -> InvestmentAnalysis: if not deals: raise ValueError("Cannot analyze investment potential from empty deals list") - # Extract price per sqm and dates + if config is None: + config = get_config() + + # Step 1: Apply outlier filtering if configured + deals_for_analysis = deals + if ( + config.analysis_use_robust_trends + and config.analysis_outlier_method != "none" + and len(deals) >= config.analysis_min_deals_for_outlier_detection + ): + filtered_deals, _ = filter_deals_for_analysis(deals, config, metric="price_per_sqm") + if filtered_deals and len(filtered_deals) >= 3: # Need at least 3 deals for analysis + deals_for_analysis = filtered_deals + logger.info( + f"Filtered {len(deals) - len(filtered_deals)} outliers " + f"from investment analysis ({len(deals)} → {len(filtered_deals)} deals)" + ) + + # Step 2: Extract price per sqm and dates from deals_for_analysis price_data = [] - for deal in deals: + for deal in deals_for_analysis: price_per_sqm = deal.price_per_sqm # Use computed field from Deal model if price_per_sqm and price_per_sqm > 0 and deal.deal_date: @@ -292,12 +319,26 @@ def analyze_investment_potential(deals: List[Deal]) -> InvestmentAnalysis: else: price_trend = "stable" - # Calculate price volatility (coefficient of variation) - std_dev = calculate_std_dev(prices) - if avg_price_per_sqm > 0: - coefficient_of_variation = (std_dev / avg_price_per_sqm) * 100 + # Step 3: Calculate price volatility + # Use robust volatility (IQR-based) if configured, otherwise use traditional CV + if config.analysis_use_robust_volatility: + # Robust volatility using IQR (less sensitive to outliers) + iqr = calculate_iqr(prices) + # Calculate median for robust CV + sorted_prices = sorted(prices) + median_price = sorted_prices[len(sorted_prices) // 2] + if median_price > 0: + # Robust coefficient of variation: IQR / median × 100 + coefficient_of_variation = (iqr / median_price) * 100 + else: + coefficient_of_variation = 0 else: - coefficient_of_variation = 0 + # Traditional volatility (coefficient of variation using std_dev) + std_dev = calculate_std_dev(prices) + if avg_price_per_sqm > 0: + coefficient_of_variation = (std_dev / avg_price_per_sqm) * 100 + else: + coefficient_of_variation = 0 # Convert CV to volatility score (0-100, lower is better) # Using defined volatility thresholds diff --git a/nadlan_mcp/govmap/models.py b/nadlan_mcp/govmap/models.py index 5632209..6ba8c9f 100644 --- a/nadlan_mcp/govmap/models.py +++ b/nadlan_mcp/govmap/models.py @@ -183,6 +183,30 @@ class Deal(BaseModel): raise TypeError(f"Unsupported type for date parsing: {type(v)}") +class OutlierReport(BaseModel): + """ + Report on outliers detected and removed from analysis. + + Provides transparency about what data was filtered and why, + allowing LLMs to understand data quality issues. + + Attributes: + total_deals: Total number of deals before filtering + outliers_removed: Number of outliers removed + outlier_indices: Indices of removed deals in original list + method_used: Outlier detection method ("iqr", "percent", "none") + parameters: Configuration parameters used for detection + """ + + total_deals: int = Field(..., description="Total deals before filtering") + outliers_removed: int = Field(..., description="Number of outliers removed") + outlier_indices: List[int] = Field(default_factory=list, description="Indices of removed deals") + method_used: str = Field(..., description="Detection method (iqr, percent, none)") + parameters: Dict[str, Any] = Field( + default_factory=dict, description="Detection parameters used" + ) + + class DealStatistics(BaseModel): """ Statistical analysis of real estate deals. @@ -222,6 +246,25 @@ class DealStatistics(BaseModel): # Date range date_range: Optional[Dict[str, str]] = Field(None, description="Earliest and latest deal dates") + # Outlier detection and filtered statistics + outlier_report: Optional[OutlierReport] = Field( + None, description="Outlier detection report (if filtering was applied)" + ) + + # Filtered statistics (after outlier removal) + filtered_deal_count: Optional[int] = Field( + None, description="Number of deals after outlier removal" + ) + filtered_price_statistics: Optional[Dict[str, float]] = Field( + None, description="Price statistics after outlier filtering" + ) + filtered_area_statistics: Optional[Dict[str, float]] = Field( + None, description="Area statistics after outlier filtering" + ) + filtered_price_per_sqm_statistics: Optional[Dict[str, float]] = Field( + None, description="Price/sqm statistics after outlier filtering" + ) + class MarketActivityScore(BaseModel): """ diff --git a/nadlan_mcp/govmap/outlier_detection.py b/nadlan_mcp/govmap/outlier_detection.py new file mode 100644 index 0000000..1f14708 --- /dev/null +++ b/nadlan_mcp/govmap/outlier_detection.py @@ -0,0 +1,315 @@ +""" +Outlier detection and robust statistical analysis for real estate data. + +This module provides functions to detect and filter outliers from deal data, +improving the accuracy of statistical analyses and market assessments. +""" + +from typing import Any, Dict, List, Optional, Tuple + +from nadlan_mcp.config import GovmapConfig, get_config +from nadlan_mcp.govmap.models import Deal + + +def calculate_iqr(values: List[float]) -> float: + """ + Calculate the Interquartile Range (IQR) of a dataset. + + Args: + values: List of numeric values + + Returns: + IQR (Q3 - Q1) + """ + if not values: + return 0.0 + + sorted_values = sorted(values) + n = len(sorted_values) + + # Calculate Q1 (25th percentile) + q1_index = n // 4 + q1 = sorted_values[q1_index] + + # Calculate Q3 (75th percentile) + q3_index = (3 * n) // 4 + q3 = sorted_values[q3_index] + + return q3 - q1 + + +def detect_outliers_iqr(values: List[float], multiplier: float = 1.5) -> List[bool]: + """ + Detect outliers using the IQR (Interquartile Range) method. + + This is the most robust method for real estate data, as it doesn't assume + a normal distribution and handles skewed data well. + + Outliers are defined as values outside [Q1 - k*IQR, Q3 + k*IQR] where: + - k=1.5 (mild outliers, more aggressive filtering) + - k=3.0 (extreme outliers, conservative filtering) + + Args: + values: List of numeric values to check + multiplier: IQR multiplier (default 1.5) + + Returns: + List of booleans, True if value is an outlier + """ + if not values or len(values) < 4: + return [False] * len(values) + + sorted_values = sorted(values) + n = len(sorted_values) + + # Calculate Q1 and Q3 + q1_index = n // 4 + q1 = sorted_values[q1_index] + + q3_index = (3 * n) // 4 + q3 = sorted_values[q3_index] + + iqr = q3 - q1 + + # Calculate bounds + lower_bound = q1 - (multiplier * iqr) + upper_bound = q3 + (multiplier * iqr) + + # Mark outliers + return [value < lower_bound or value > upper_bound for value in values] + + +def detect_outliers_percent(values: List[float], threshold: float = 0.5) -> List[bool]: + """ + Detect outliers using percentage-based thresholds from the median. + + Outliers are defined as values outside [median * (1-threshold), median * (1+threshold)] + + Example: threshold=0.5 means values >150% or <50% of median are outliers + + Args: + values: List of numeric values to check + threshold: Percentage threshold (0.5 = 50%) + + Returns: + List of booleans, True if value is an outlier + """ + if not values: + return [] + + sorted_values = sorted(values) + n = len(sorted_values) + median = sorted_values[n // 2] + + lower_bound = median * (1 - threshold) + upper_bound = median * (1 + threshold) + + return [value < lower_bound or value > upper_bound for value in values] + + +def apply_hard_bounds_price_per_sqm( + deals: List[Deal], config: Optional[GovmapConfig] = None +) -> List[bool]: + """ + Apply hard bounds filtering to price per sqm values. + + This catches obvious data entry errors (e.g., wrong area leading to + extreme price per sqm values). + + Args: + deals: List of deals to check + config: Configuration object (optional, uses global if not provided) + + Returns: + List of booleans, True if deal should be filtered out + """ + if config is None: + config = get_config() + + filters = [] + for deal in deals: + price_per_sqm = deal.price_per_sqm + + # Skip deals without valid price per sqm + if price_per_sqm is None: + filters.append(False) + continue + + # Check bounds + is_outlier = ( + price_per_sqm < config.analysis_price_per_sqm_min + or price_per_sqm > config.analysis_price_per_sqm_max + ) + filters.append(is_outlier) + + return filters + + +def apply_hard_bounds_deal_amount( + deals: List[Deal], config: Optional[GovmapConfig] = None +) -> List[bool]: + """ + Apply hard bounds filtering to deal amounts. + + This catches partial deals and obvious data errors (e.g., 400K apartment + when typical apartments are 1.6M). + + Args: + deals: List of deals to check + config: Configuration object (optional, uses global if not provided) + + Returns: + List of booleans, True if deal should be filtered out + """ + if config is None: + config = get_config() + + filters = [] + for deal in deals: + is_outlier = deal.deal_amount < config.analysis_min_deal_amount + filters.append(is_outlier) + + return filters + + +def filter_deals_for_analysis( + deals: List[Deal], config: Optional[GovmapConfig] = None, metric: str = "price_per_sqm" +) -> Tuple[List[Deal], Dict[str, Any]]: + """ + Filter deals to remove outliers based on configuration. + + This is the main entry point for outlier filtering. It applies a combination + of hard bounds and statistical outlier detection. + + Process: + 1. Apply hard bounds to price per sqm (catches data errors) + 2. Apply hard bounds to deal amount (catches partial deals) + 3. Apply statistical outlier detection (IQR or percent method) to specified metric + 4. Combine all filters and remove outliers + + Args: + deals: List of deals to filter + config: Configuration object (optional, uses global if not provided) + metric: Which metric to apply statistical outlier detection to + Options: "price_per_sqm", "deal_amount" + + Returns: + Tuple of: + - Filtered list of deals + - Outlier report dict with details on what was filtered + """ + if config is None: + config = get_config() + + if not deals: + return [], { + "total_deals": 0, + "outliers_removed": 0, + "outlier_indices": [], + "method_used": config.analysis_outlier_method, + "parameters": {}, + } + + # Skip outlier detection if disabled or insufficient data + if ( + config.analysis_outlier_method == "none" + or len(deals) < config.analysis_min_deals_for_outlier_detection + ): + return deals, { + "total_deals": len(deals), + "outliers_removed": 0, + "outlier_indices": [], + "method_used": "none", + "parameters": {"reason": "disabled or insufficient data"}, + } + + # Initialize filter masks (True = keep, False = remove) + filters_to_remove = [False] * len(deals) + + # Step 1: Apply hard bounds to price per sqm + price_per_sqm_outliers = apply_hard_bounds_price_per_sqm(deals, config) + for i, is_outlier in enumerate(price_per_sqm_outliers): + if is_outlier: + filters_to_remove[i] = True + + # Step 2: Apply hard bounds to deal amount + deal_amount_outliers = apply_hard_bounds_deal_amount(deals, config) + for i, is_outlier in enumerate(deal_amount_outliers): + if is_outlier: + filters_to_remove[i] = True + + # Step 3: Apply statistical outlier detection to specified metric + if config.analysis_outlier_method == "iqr": + # Extract values for the specified metric + if metric == "price_per_sqm": + values = [ + deal.price_per_sqm + for deal in deals + if deal.price_per_sqm is not None and not filters_to_remove[i] + ] + value_indices = [ + i + for i, deal in enumerate(deals) + if deal.price_per_sqm is not None and not filters_to_remove[i] + ] + elif metric == "deal_amount": + values = [deal.deal_amount for i, deal in enumerate(deals) if not filters_to_remove[i]] + value_indices = [i for i in range(len(deals)) if not filters_to_remove[i]] + else: + values = [] + value_indices = [] + + if values: + statistical_outliers = detect_outliers_iqr(values, config.analysis_iqr_multiplier) + for i, is_outlier in enumerate(statistical_outliers): + if is_outlier: + filters_to_remove[value_indices[i]] = True + + elif config.analysis_outlier_method == "percent": + # Extract values for the specified metric + if metric == "price_per_sqm": + values = [ + deal.price_per_sqm + for i, deal in enumerate(deals) + if deal.price_per_sqm is not None and not filters_to_remove[i] + ] + value_indices = [ + i + for i, deal in enumerate(deals) + if deal.price_per_sqm is not None and not filters_to_remove[i] + ] + elif metric == "deal_amount": + values = [deal.deal_amount for i, deal in enumerate(deals) if not filters_to_remove[i]] + value_indices = [i for i in range(len(deals)) if not filters_to_remove[i]] + else: + values = [] + value_indices = [] + + if values: + statistical_outliers = detect_outliers_percent(values, 0.5) + for i, is_outlier in enumerate(statistical_outliers): + if is_outlier: + filters_to_remove[value_indices[i]] = True + + # Filter deals + filtered_deals = [deal for i, deal in enumerate(deals) if not filters_to_remove[i]] + outlier_indices = [i for i, should_remove in enumerate(filters_to_remove) if should_remove] + + # Create outlier report + report = { + "total_deals": len(deals), + "outliers_removed": len(outlier_indices), + "outlier_indices": outlier_indices, + "method_used": config.analysis_outlier_method, + "parameters": { + "iqr_multiplier": config.analysis_iqr_multiplier + if config.analysis_outlier_method == "iqr" + else None, + "metric": metric, + "price_per_sqm_min": config.analysis_price_per_sqm_min, + "price_per_sqm_max": config.analysis_price_per_sqm_max, + "min_deal_amount": config.analysis_min_deal_amount, + }, + } + + return filtered_deals, report diff --git a/nadlan_mcp/govmap/statistics.py b/nadlan_mcp/govmap/statistics.py index 326a7ec..4076161 100644 --- a/nadlan_mcp/govmap/statistics.py +++ b/nadlan_mcp/govmap/statistics.py @@ -6,39 +6,27 @@ This module provides pure mathematical functions for analyzing real estate deal from collections import Counter import logging -from typing import List +from typing import Dict, List, Optional -from .models import Deal, DealStatistics +from nadlan_mcp.config import GovmapConfig, get_config + +from .models import Deal, DealStatistics, OutlierReport +from .outlier_detection import filter_deals_for_analysis logger = logging.getLogger(__name__) -def calculate_deal_statistics(deals: List[Deal]) -> DealStatistics: +def _calculate_basic_stats(deals: List[Deal]) -> Dict: """ - Calculate statistical aggregations on deal data. + Internal helper to calculate basic statistics from a deal list. Args: deals: List of Deal model instances Returns: - DealStatistics model with comprehensive metrics - - Raises: - ValueError: If deals is not a valid list + Dictionary with price_stats, area_stats, price_per_sqm_stats, + property_type_dist, and date_range """ - if not isinstance(deals, list): - raise ValueError("deals must be a list") - - if not deals: - return DealStatistics( - total_deals=0, - price_statistics={}, - area_statistics={}, - price_per_sqm_statistics={}, - property_type_distribution={}, - date_range=None, - ) - # Extract numeric values prices = [] areas = [] @@ -154,13 +142,91 @@ def calculate_deal_statistics(deals: List[Deal]) -> DealStatistics: logger.warning("Invalid date format in date range calculation") pass + return { + "price_statistics": price_stats, + "area_statistics": area_stats, + "price_per_sqm_statistics": price_per_sqm_stats, + "property_type_distribution": property_type_dist, + "date_range": date_range_dict, + } + + +def calculate_deal_statistics( + deals: List[Deal], config: Optional[GovmapConfig] = None +) -> DealStatistics: + """ + Calculate statistical aggregations on deal data with optional outlier filtering. + + This function calculates comprehensive statistics on real estate deals, optionally + filtering outliers based on configuration. When outlier filtering is enabled, it + returns both original (unfiltered) and filtered statistics for transparency. + + Args: + deals: List of Deal model instances + config: Configuration object (optional, uses global config if not provided) + + Returns: + DealStatistics model with comprehensive metrics, including: + - Original statistics (calculated on all deals) + - Filtered statistics (calculated after outlier removal, if enabled) + - Outlier report (details on what was filtered, if enabled) + + Raises: + ValueError: If deals is not a valid list + """ + if not isinstance(deals, list): + raise ValueError("deals must be a list") + + if config is None: + config = get_config() + + if not deals: + return DealStatistics( + total_deals=0, + price_statistics={}, + area_statistics={}, + price_per_sqm_statistics={}, + property_type_distribution={}, + date_range=None, + ) + + # Step 1: Calculate statistics on original data + original_stats = _calculate_basic_stats(deals) + + # Step 2: Apply outlier filtering if enabled + outlier_report_data = None + filtered_stats = None + + if ( + config.analysis_outlier_method != "none" + and len(deals) >= config.analysis_min_deals_for_outlier_detection + ): + # Filter deals for analysis (primarily targeting price_per_sqm outliers) + filtered_deals, report_dict = filter_deals_for_analysis( + deals, config, metric="price_per_sqm" + ) + + # Create OutlierReport model + outlier_report_data = OutlierReport(**report_dict) + + # Calculate statistics on filtered data + if filtered_deals and len(filtered_deals) > 0: + filtered_basic_stats = _calculate_basic_stats(filtered_deals) + filtered_stats = { + "filtered_deal_count": len(filtered_deals), + "filtered_price_statistics": filtered_basic_stats["price_statistics"], + "filtered_area_statistics": filtered_basic_stats["area_statistics"], + "filtered_price_per_sqm_statistics": filtered_basic_stats[ + "price_per_sqm_statistics" + ], + } + + # Step 3: Return comprehensive DealStatistics with both original and filtered data return DealStatistics( total_deals=len(deals), - price_statistics=price_stats, - area_statistics=area_stats, - price_per_sqm_statistics=price_per_sqm_stats, - property_type_distribution=property_type_dist, - date_range=date_range_dict, + **original_stats, + outlier_report=outlier_report_data, + **(filtered_stats if filtered_stats else {}), ) diff --git a/tests/govmap/test_outlier_detection.py b/tests/govmap/test_outlier_detection.py new file mode 100644 index 0000000..1e64d87 --- /dev/null +++ b/tests/govmap/test_outlier_detection.py @@ -0,0 +1,331 @@ +""" +Tests for outlier detection functions. +""" + +from datetime import date + +from nadlan_mcp.config import GovmapConfig +from nadlan_mcp.govmap.models import Deal +from nadlan_mcp.govmap.outlier_detection import ( + apply_hard_bounds_deal_amount, + apply_hard_bounds_price_per_sqm, + calculate_iqr, + detect_outliers_iqr, + detect_outliers_percent, + filter_deals_for_analysis, +) + + +class TestCalculateIQR: + """Tests for calculate_iqr function.""" + + def test_calculate_iqr_normal_data(self): + """Test IQR calculation with normal data.""" + values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] + iqr = calculate_iqr(values) + # Q1 = 3 (index 2), Q3 = 8 (index 7), IQR = 5 + assert iqr == 5 + + def test_calculate_iqr_empty_list(self): + """Test IQR calculation with empty list.""" + assert calculate_iqr([]) == 0.0 + + def test_calculate_iqr_single_value(self): + """Test IQR calculation with single value.""" + assert calculate_iqr([5.0]) == 0.0 + + +class TestDetectOutliersIQR: + """Tests for detect_outliers_iqr function.""" + + def test_detect_outliers_iqr_with_outliers(self): + """Test IQR outlier detection with clear outliers.""" + # Dataset: 1-10 with outliers at 50 and 100 + values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 50, 100] + outliers = detect_outliers_iqr(values, multiplier=1.5) + + # The last two values (50, 100) should be outliers + assert outliers[-2] is True # 50 + assert outliers[-1] is True # 100 + # The first 10 values should not be outliers + assert sum(outliers[:10]) == 0 + + def test_detect_outliers_iqr_no_outliers(self): + """Test IQR outlier detection with no outliers.""" + values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] + outliers = detect_outliers_iqr(values, multiplier=1.5) + assert sum(outliers) == 0 + + def test_detect_outliers_iqr_empty_list(self): + """Test IQR outlier detection with empty list.""" + assert detect_outliers_iqr([]) == [] + + def test_detect_outliers_iqr_insufficient_data(self): + """Test IQR outlier detection with insufficient data.""" + values = [1, 2, 3] + outliers = detect_outliers_iqr(values, multiplier=1.5) + assert outliers == [False, False, False] + + def test_detect_outliers_iqr_different_multipliers(self): + """Test IQR outlier detection with different multipliers.""" + values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 50] + + # More aggressive (k=1.5) + outliers_aggressive = detect_outliers_iqr(values, multiplier=1.5) + # Conservative (k=3.0) + outliers_conservative = detect_outliers_iqr(values, multiplier=3.0) + + # Aggressive should detect more outliers + assert sum(outliers_aggressive) >= sum(outliers_conservative) + + +class TestDetectOutliersPercent: + """Tests for detect_outliers_percent function.""" + + def test_detect_outliers_percent_with_outliers(self): + """Test percent-based outlier detection with clear outliers.""" + # Median = 10, threshold=0.5 means 5-15 is acceptable, <5 or >15 is outlier + values = [2, 9, 10, 10, 10, 11, 20] + outliers = detect_outliers_percent(values, threshold=0.5) + + assert outliers[0] is True # 2 < 5 + assert outliers[-1] is True # 20 > 15 + # Middle values should not be outliers + assert sum(outliers[1:-1]) == 0 + + def test_detect_outliers_percent_no_outliers(self): + """Test percent-based outlier detection with no outliers.""" + values = [8, 9, 10, 11, 12] + outliers = detect_outliers_percent(values, threshold=0.5) + assert sum(outliers) == 0 + + def test_detect_outliers_percent_empty_list(self): + """Test percent-based outlier detection with empty list.""" + assert detect_outliers_percent([]) == [] + + +class TestApplyHardBoundsPricePerSqm: + """Tests for apply_hard_bounds_price_per_sqm function.""" + + def test_hard_bounds_price_per_sqm_within_bounds(self): + """Test hard bounds filtering with values within bounds.""" + deals = [ + Deal( + objectid=1, + deal_amount=1000000, + deal_date=date(2023, 1, 1), + asset_area=100, # price_per_sqm = 10000 + ), + Deal( + objectid=2, + deal_amount=2000000, + deal_date=date(2023, 1, 1), + asset_area=100, # price_per_sqm = 20000 + ), + ] + config = GovmapConfig() + filters = apply_hard_bounds_price_per_sqm(deals, config) + assert filters == [False, False] # None should be filtered + + def test_hard_bounds_price_per_sqm_outside_bounds(self): + """Test hard bounds filtering with values outside bounds.""" + deals = [ + Deal( + objectid=1, + deal_amount=500, + deal_date=date(2023, 1, 1), + asset_area=1, # price_per_sqm = 500 (< 1000 min) + ), + Deal( + objectid=2, + deal_amount=10000000, + deal_date=date(2023, 1, 1), + asset_area=10, # price_per_sqm = 1000000 (> 100000 max) + ), + ] + config = GovmapConfig() + filters = apply_hard_bounds_price_per_sqm(deals, config) + assert filters == [True, True] # Both should be filtered + + def test_hard_bounds_price_per_sqm_no_area(self): + """Test hard bounds filtering with missing area data.""" + deals = [ + Deal(objectid=1, deal_amount=1000000, deal_date=date(2023, 1, 1)), # No area + ] + config = GovmapConfig() + filters = apply_hard_bounds_price_per_sqm(deals, config) + assert filters == [False] # Not filtered (can't calculate price_per_sqm) + + +class TestApplyHardBoundsDealAmount: + """Tests for apply_hard_bounds_deal_amount function.""" + + def test_hard_bounds_deal_amount_within_bounds(self): + """Test hard bounds filtering for deal amounts within bounds.""" + deals = [ + Deal(objectid=1, deal_amount=500000, deal_date=date(2023, 1, 1)), # > 100K (OK) + Deal(objectid=2, deal_amount=1000000, deal_date=date(2023, 1, 1)), # > 100K (OK) + ] + config = GovmapConfig() + filters = apply_hard_bounds_deal_amount(deals, config) + assert filters == [False, False] # Both are >= 100K, so not filtered + + def test_hard_bounds_deal_amount_below_minimum(self): + """Test hard bounds filtering for deal amounts below minimum.""" + deals = [ + Deal(objectid=1, deal_amount=50000, deal_date=date(2023, 1, 1)), # Below 100K + Deal(objectid=2, deal_amount=100000, deal_date=date(2023, 1, 1)), # At 100K (OK) + ] + config = GovmapConfig() + filters = apply_hard_bounds_deal_amount(deals, config) + assert filters == [True, False] + + +class TestFilterDealsForAnalysis: + """Tests for filter_deals_for_analysis function.""" + + def test_filter_deals_empty_list(self): + """Test filtering with empty deal list.""" + config = GovmapConfig() + filtered, report = filter_deals_for_analysis([], config) + assert filtered == [] + assert report["total_deals"] == 0 + assert report["outliers_removed"] == 0 + + def test_filter_deals_disabled(self): + """Test filtering with outlier detection disabled.""" + deals = [ + Deal(objectid=1, deal_amount=1000000, deal_date=date(2023, 1, 1), asset_area=100), + Deal(objectid=2, deal_amount=2000000, deal_date=date(2023, 1, 1), asset_area=100), + ] + config = GovmapConfig(analysis_outlier_method="none") + filtered, report = filter_deals_for_analysis(deals, config) + assert len(filtered) == 2 + assert report["method_used"] == "none" + assert report["outliers_removed"] == 0 + + def test_filter_deals_insufficient_data(self): + """Test filtering with insufficient data for outlier detection.""" + deals = [ + Deal(objectid=1, deal_amount=1000000, deal_date=date(2023, 1, 1), asset_area=100), + Deal(objectid=2, deal_amount=2000000, deal_date=date(2023, 1, 1), asset_area=100), + ] + config = GovmapConfig(analysis_min_deals_for_outlier_detection=10) + filtered, report = filter_deals_for_analysis(deals, config) + assert len(filtered) == 2 + assert report["outliers_removed"] == 0 + + def test_filter_deals_with_outliers(self): + """Test filtering with actual outliers in data.""" + # Create deals with one clear outlier (very low price_per_sqm) + deals = [] + for i in range(15): + deals.append( + Deal( + objectid=i, + deal_amount=1000000, + deal_date=date(2023, 1, 1), + asset_area=100, # Normal: 10K/sqm + ) + ) + # Add outlier: very high price_per_sqm + deals.append( + Deal( + objectid=100, + deal_amount=10000000, + deal_date=date(2023, 1, 1), + asset_area=10, # Outlier: 1M/sqm (way too high) + ) + ) + + config = GovmapConfig(analysis_outlier_method="iqr", analysis_iqr_multiplier=1.5) + filtered, report = filter_deals_for_analysis(deals, config, metric="price_per_sqm") + + assert len(filtered) < len(deals) # Some deals should be filtered + assert report["outliers_removed"] > 0 + assert report["method_used"] == "iqr" + assert ( + 15 in report["outlier_indices"] + ) # The outlier deal (at index 15) should be in the list + + def test_filter_deals_hard_bounds_only(self): + """Test filtering with hard bounds catching outliers.""" + deals = [ + # Normal deals + Deal(objectid=1, deal_amount=1000000, deal_date=date(2023, 1, 1), asset_area=100), + Deal(objectid=2, deal_amount=1100000, deal_date=date(2023, 1, 1), asset_area=100), + # Data error: price_per_sqm way too high (2M/sqm > 100K max) + Deal(objectid=3, deal_amount=10000000, deal_date=date(2023, 1, 1), asset_area=5), + ] + # Need to set min_deals to 1 so filtering actually runs (default is 10) + config = GovmapConfig(analysis_min_deals_for_outlier_detection=1) + filtered, report = filter_deals_for_analysis(deals, config) + + # Should filter the last deal due to hard bounds (price_per_sqm = 2M > 100K) + assert len(filtered) == 2 + assert report["outliers_removed"] == 1 + + def test_filter_deals_percent_method(self): + """Test filtering using percent-based method.""" + # Create deals with clear outliers + deals = [] + for i in range(12): + deals.append( + Deal(objectid=i, deal_amount=1000000, deal_date=date(2023, 1, 1), asset_area=100) + ) + # Add outlier + deals.append( + Deal(objectid=100, deal_amount=5000000, deal_date=date(2023, 1, 1), asset_area=100) + ) + + config = GovmapConfig(analysis_outlier_method="percent") + filtered, report = filter_deals_for_analysis(deals, config, metric="price_per_sqm") + + assert report["method_used"] == "percent" + # Outlier might be filtered depending on the threshold + assert len(filtered) <= len(deals) + + +class TestIntegration: + """Integration tests combining multiple functions.""" + + def test_real_world_scenario_partial_deal(self): + """Test filtering a real-world scenario with a partial deal.""" + # Scenario: 15 normal apartments around 1.6M, one partial deal at 400K + deals = [] + for i in range(15): + deals.append( + Deal(objectid=i, deal_amount=1600000, deal_date=date(2023, 1, 1), asset_area=100) + ) + # Partial deal (outlier) + deals.append( + Deal(objectid=100, deal_amount=400000, deal_date=date(2023, 1, 1), asset_area=100) + ) + + config = GovmapConfig(analysis_outlier_method="iqr", analysis_iqr_multiplier=1.5) + filtered, report = filter_deals_for_analysis(deals, config, metric="price_per_sqm") + + # The partial deal should be filtered out + assert len(filtered) == 15 + assert report["outliers_removed"] == 1 + assert 15 in report["outlier_indices"] # Index 15 (not objectid 100) + + def test_real_world_scenario_data_entry_error(self): + """Test filtering a data entry error (wrong area).""" + # Scenario: Normal deals + one with wrong area (1 sqm instead of 100) + deals = [] + for i in range(12): + deals.append( + Deal(objectid=i, deal_amount=1000000, deal_date=date(2023, 1, 1), asset_area=100) + ) + # Data entry error: area=1 instead of 100 + deals.append( + Deal(objectid=100, deal_amount=1000000, deal_date=date(2023, 1, 1), asset_area=1) + ) + + config = GovmapConfig() + filtered, report = filter_deals_for_analysis(deals, config) + + # The data error should be filtered by hard bounds (price_per_sqm > 100K) + assert len(filtered) == 12 + assert report["outliers_removed"] == 1