From 7757694077832c146e1ae315f6af6e5a387db5b4 Mon Sep 17 00:00:00 2001 From: Nitzan Pomerantz <9297302+nitzpo@users.noreply.github.com> Date: Sat, 25 Oct 2025 13:18:19 +0300 Subject: [PATCH] Phase 3: Complete govmap package extraction (step 2/3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Created market_analysis.py with market analysis functions: - parse_deal_dates() - Date parsing and filtering helper - calculate_market_activity_score() - Activity metrics - analyze_investment_potential() - Investment analysis - get_market_liquidity() - Liquidity metrics - Created client.py with GovmapClient class: - Core API methods (autocomplete, get deals, etc.) - Validation methods delegating to validators module - Utility methods delegating to utils module - Filtering, statistics, and analysis methods delegating to respective modules - Updated govmap/__init__.py to export GovmapClient All modules maintain backward compatibility through delegation pattern. Next: Delete old govmap.py and update imports. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- nadlan_mcp/govmap/__init__.py | 66 +++ nadlan_mcp/govmap/client.py | 740 +++++++++++++++++++++++++++ nadlan_mcp/govmap/market_analysis.py | 422 +++++++++++++++ 3 files changed, 1228 insertions(+) create mode 100644 nadlan_mcp/govmap/__init__.py create mode 100644 nadlan_mcp/govmap/client.py create mode 100644 nadlan_mcp/govmap/market_analysis.py diff --git a/nadlan_mcp/govmap/__init__.py b/nadlan_mcp/govmap/__init__.py new file mode 100644 index 0000000..fcddc5a --- /dev/null +++ b/nadlan_mcp/govmap/__init__.py @@ -0,0 +1,66 @@ +""" +Govmap package - Israeli government real estate data API client. + +This package provides a modular interface to the Govmap API for querying +Israeli real estate deals, market trends, and property information. + +Public API: + - GovmapClient: Main API client class (to be added from client.py) + - filter_deals_by_criteria: Filter deals by various criteria + - calculate_deal_statistics: Calculate statistical aggregations + - calculate_market_activity_score: Market activity and trend metrics + - analyze_investment_potential: Investment analysis and price trends + - get_market_liquidity: Market liquidity and velocity metrics +""" + +# Filter functions +from .filters import filter_deals_by_criteria + +# Statistics functions +from .statistics import calculate_deal_statistics, calculate_std_dev + +# Market analysis functions +from .market_analysis import ( + calculate_market_activity_score, + analyze_investment_potential, + get_market_liquidity, + parse_deal_dates, +) + +# Utility functions +from .utils import calculate_distance, is_same_building, extract_floor_number + +# Validation functions +from .validators import ( + validate_address, + validate_coordinates, + validate_positive_int, + validate_deal_type, +) + +# Main API client +from .client import GovmapClient + +__all__ = [ + # Main client class + "GovmapClient", + # Filtering + "filter_deals_by_criteria", + # Statistics + "calculate_deal_statistics", + "calculate_std_dev", + # Market analysis + "calculate_market_activity_score", + "analyze_investment_potential", + "get_market_liquidity", + "parse_deal_dates", + # Utilities + "calculate_distance", + "is_same_building", + "extract_floor_number", + # Validation + "validate_address", + "validate_coordinates", + "validate_positive_int", + "validate_deal_type", +] diff --git a/nadlan_mcp/govmap/client.py b/nadlan_mcp/govmap/client.py new file mode 100644 index 0000000..44320c8 --- /dev/null +++ b/nadlan_mcp/govmap/client.py @@ -0,0 +1,740 @@ +""" +Govmap API Client for Israeli real estate data. + +This module provides the main GovmapClient class for interacting with the +Israeli government's Govmap API to retrieve property deals, market trends, +and real estate information. +""" + +import logging +import time +from typing import Any, Dict, List, Optional, Tuple +from datetime import datetime, timedelta + +import requests + +from nadlan_mcp.config import GovmapConfig, get_config + +# Import functions from modular package +from . import validators +from . import utils +from . import filters +from . import statistics +from . import market_analysis + +logger = logging.getLogger(__name__) + + +class GovmapClient: + """ + A client for interacting with the Israeli government's Govmap API. + + This class provides methods to search for properties, find block/parcel information, + and retrieve real estate deal data with automatic retries and rate limiting. + + Attributes: + config: Configuration object with API settings + session: Requests session for connection pooling + last_request_time: Timestamp of last API request for rate limiting + """ + + def __init__(self, config: Optional[GovmapConfig] = None): + """ + Initialize the GovmapClient. + + Args: + config: Optional configuration object. If None, uses global config. + """ + self.config = config or get_config() + self.base_url = self.config.base_url.rstrip("/") + self.session = requests.Session() + self.session.headers.update( + {"Content-Type": "application/json", "User-Agent": self.config.user_agent} + ) + self.last_request_time = 0.0 + + def _rate_limit(self): + """ + Enforce rate limiting by sleeping if necessary. + + Ensures requests don't exceed the configured requests_per_second. + """ + min_interval = 1.0 / self.config.requests_per_second + elapsed = time.time() - self.last_request_time + if elapsed < min_interval: + time.sleep(min_interval - elapsed) + self.last_request_time = time.time() + + # Validation methods (delegate to validators module) + def _validate_address(self, address: str) -> str: + """Validate and sanitize address input.""" + return validators.validate_address(address) + + def _validate_coordinates(self, point: Tuple[float, float]) -> Tuple[float, float]: + """Validate coordinate input.""" + return validators.validate_coordinates(point) + + def _validate_positive_int( + self, value: int, name: str, max_value: Optional[int] = None + ) -> int: + """Validate positive integer input.""" + return validators.validate_positive_int(value, name, max_value) + + # Utility methods (delegate to utils module) + def _calculate_distance( + self, point1: Tuple[float, float], point2: Tuple[float, float] + ) -> float: + """Calculate Euclidean distance between two points in ITM coordinates.""" + return utils.calculate_distance(point1, point2) + + def _is_same_building(self, search_address: str, deal_address: str) -> bool: + """Check if a deal is from the same building as the search address.""" + return utils.is_same_building(search_address, deal_address) + + def _extract_floor_number(self, floor_str: str) -> Optional[int]: + """Extract numeric floor number from Hebrew floor description.""" + return utils.extract_floor_number(floor_str) + + # Core API methods + def autocomplete_address(self, search_text: str) -> Dict[str, Any]: + """ + Find the most likely match for a given address using autocomplete. + + Args: + search_text: The address to search for (e.g., "סוקולוב 38 חולון") + + Returns: + Dict containing the JSON response from the API with coordinates + + Raises: + requests.RequestException: If the API request fails after retries + ValueError: If the response is invalid or input is invalid + """ + search_text = self._validate_address(search_text) + url = f"{self.base_url}/search-service/autocomplete" + + payload = { + "searchText": search_text, + "language": "he", + "isAccurate": False, + "maxResults": 10, + } + + # Retry logic with exponential backoff + for attempt in range(self.config.max_retries + 1): + try: + self._rate_limit() + + logger.info( + f"Searching for address: {search_text} (attempt {attempt + 1}/{self.config.max_retries + 1})" + ) + timeout = (self.config.connect_timeout, self.config.read_timeout) + response = self.session.post(url, json=payload, timeout=timeout) + response.raise_for_status() + + data = response.json() + if not data or "results" not in data: + raise ValueError("Invalid response format from autocomplete API") + + return data + + except (requests.RequestException, requests.Timeout) as e: + if attempt < self.config.max_retries: + wait_time = min( + self.config.retry_min_wait * (2**attempt), + self.config.retry_max_wait, + ) + logger.warning( + f"Request failed (attempt {attempt + 1}), retrying in {wait_time}s: {e}" + ) + time.sleep(wait_time) + else: + logger.error( + f"Request failed after {self.config.max_retries + 1} attempts: {e}" + ) + raise + # This line should never be reached but satisfies type checker + raise RuntimeError( + "Unexpected error: retry loop exited without return or raise" + ) + + def get_gush_helka(self, point: Tuple[float, float]) -> Dict[str, Any]: + """ + Get Gush (Block) and Helka (Parcel) information for a coordinate point. + + Args: + point: A tuple of (longitude, latitude) + + Returns: + Dict containing the JSON response with block and parcel data + + Raises: + requests.RequestException: If the API request fails after retries + ValueError: If the response or input is invalid + """ + point = self._validate_coordinates(point) + url = f"{self.base_url}/layers-catalog/entitiesByPoint" + + payload = {"point": list(point), "layers": [{"layerId": "16"}], "tolerance": 0} + + # Retry logic with exponential backoff + for attempt in range(self.config.max_retries + 1): + try: + self._rate_limit() + + logger.info( + f"Getting Gush/Helka for point: {point} (attempt {attempt + 1}/{self.config.max_retries + 1})" + ) + timeout = (self.config.connect_timeout, self.config.read_timeout) + response = self.session.post(url, json=payload, timeout=timeout) + response.raise_for_status() + + data = response.json() + return data + + except (requests.RequestException, requests.Timeout) as e: + if attempt < self.config.max_retries: + wait_time = min( + self.config.retry_min_wait * (2**attempt), + self.config.retry_max_wait, + ) + logger.warning( + f"Request failed (attempt {attempt + 1}), retrying in {wait_time}s: {e}" + ) + time.sleep(wait_time) + else: + logger.error( + f"Request failed after {self.config.max_retries + 1} attempts: {e}" + ) + raise + # This line should never be reached but satisfies type checker + raise RuntimeError( + "Unexpected error: retry loop exited without return or raise" + ) + + def get_deals_by_radius( + self, point: Tuple[float, float], radius: int = 50 + ) -> List[Dict[str, Any]]: + """ + Find real estate deals within a specified radius of a point. + + Args: + point: A tuple of (longitude, latitude) + radius: The search radius in meters (default: 50) + + Returns: + List of deals found within the radius + + Raises: + requests.RequestException: If the API request fails after retries + ValueError: If the response or input is invalid + """ + point = self._validate_coordinates(point) + radius = self._validate_positive_int(radius, "radius", max_value=5000) + url = f"{self.base_url}/real-estate/deals/{point[0]},{point[1]}/{radius}" + + # Retry logic with exponential backoff + for attempt in range(self.config.max_retries + 1): + try: + self._rate_limit() + + logger.info( + f"Getting deals by radius for point: {point}, radius: {radius}m (attempt {attempt + 1}/{self.config.max_retries + 1})" + ) + timeout = (self.config.connect_timeout, self.config.read_timeout) + response = self.session.get(url, timeout=timeout) + response.raise_for_status() + + data = response.json() + if not isinstance(data, list): + raise ValueError( + f"Expected list response, got {type(data).__name__}" + ) + return data + + except (requests.RequestException, requests.Timeout) as e: + if attempt < self.config.max_retries: + wait_time = min( + self.config.retry_min_wait * (2**attempt), + self.config.retry_max_wait, + ) + logger.warning( + f"Request failed (attempt {attempt + 1}), retrying in {wait_time}s: {e}" + ) + time.sleep(wait_time) + else: + logger.error( + f"Request failed after {self.config.max_retries + 1} attempts: {e}" + ) + raise + # This line should never be reached but satisfies type checker + raise RuntimeError( + "Unexpected error: retry loop exited without return or raise" + ) + + def get_street_deals( + self, + polygon_id: str, + limit: int = 10, + start_date: Optional[str] = None, + end_date: Optional[str] = None, + deal_type: int = 2, + ) -> List[Dict[str, Any]]: + """ + Retrieve detailed information about deals on a specific street. + + Args: + polygon_id: The ID of the lot's polygon + limit: Maximum number of deals to return (default: 10) + start_date: Start date for search in 'YYYY-MM' format + end_date: End date for search in 'YYYY-MM' format + deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) + + Returns: + List of detailed deal information for the street + + Raises: + requests.RequestException: If the API request fails after retries + ValueError: If the response or input is invalid + """ + if not polygon_id or not isinstance(polygon_id, str): + raise ValueError("polygon_id must be a non-empty string") + polygon_id = polygon_id.strip() + if not polygon_id: + raise ValueError("polygon_id cannot be empty or whitespace only") + + limit = self._validate_positive_int(limit, "limit", max_value=1000) + validators.validate_deal_type(deal_type) + + url = f"{self.base_url}/real-estate/street-deals/{polygon_id}" + + params: Dict[str, Any] = {"limit": limit, "dealType": deal_type} + if start_date: + params["startDate"] = start_date + if end_date: + params["endDate"] = end_date + + # Retry logic with exponential backoff + for attempt in range(self.config.max_retries + 1): + try: + self._rate_limit() + + logger.info( + f"Getting street deals for polygon: {polygon_id}, dealType: {deal_type} (attempt {attempt + 1}/{self.config.max_retries + 1})" + ) + timeout = (self.config.connect_timeout, self.config.read_timeout) + response = self.session.get(url, params=params, timeout=timeout) + response.raise_for_status() + + data = response.json() + # API returns {data: [...], totalCount: ..., limit: ..., offset: ...} + if isinstance(data, dict) and "data" in data: + if not isinstance(data["data"], list): + raise ValueError( + f"Expected list in 'data' field, got {type(data['data']).__name__}" + ) + return data["data"] + elif isinstance(data, list): + return data + else: + raise ValueError( + f"Unexpected response format: {type(data).__name__}" + ) + + except (requests.RequestException, requests.Timeout) as e: + if attempt < self.config.max_retries: + wait_time = min( + self.config.retry_min_wait * (2**attempt), + self.config.retry_max_wait, + ) + logger.warning( + f"Request failed (attempt {attempt + 1}), retrying in {wait_time}s: {e}" + ) + time.sleep(wait_time) + else: + logger.error( + f"Request failed after {self.config.max_retries + 1} attempts: {e}" + ) + raise + # This line should never be reached but satisfies type checker + raise RuntimeError( + "Unexpected error: retry loop exited without return or raise" + ) + + def get_neighborhood_deals( + self, + polygon_id: str, + limit: int = 10, + start_date: Optional[str] = None, + end_date: Optional[str] = None, + deal_type: int = 2, + ) -> List[Dict[str, Any]]: + """ + Retrieve deals within the same neighborhood as the given polygon_id. + + Args: + polygon_id: The ID of the lot's polygon + limit: Maximum number of deals to return (default: 10) + start_date: Start date for search in 'YYYY-MM' format + end_date: End date for search in 'YYYY-MM' format + deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) + + Returns: + List of deals in the neighborhood + + Raises: + requests.RequestException: If the API request fails after retries + ValueError: If the response or input is invalid + """ + if not polygon_id or not isinstance(polygon_id, str): + raise ValueError("polygon_id must be a non-empty string") + polygon_id = polygon_id.strip() + if not polygon_id: + raise ValueError("polygon_id cannot be empty or whitespace only") + + limit = self._validate_positive_int(limit, "limit", max_value=1000) + validators.validate_deal_type(deal_type) + + url = f"{self.base_url}/real-estate/neighborhood-deals/{polygon_id}" + + params: Dict[str, Any] = {"limit": limit, "dealType": deal_type} + if start_date: + params["startDate"] = start_date + if end_date: + params["endDate"] = end_date + + # Retry logic with exponential backoff + for attempt in range(self.config.max_retries + 1): + try: + self._rate_limit() + + logger.info( + f"Getting neighborhood deals for polygon: {polygon_id}, dealType: {deal_type} (attempt {attempt + 1}/{self.config.max_retries + 1})" + ) + timeout = (self.config.connect_timeout, self.config.read_timeout) + response = self.session.get(url, params=params, timeout=timeout) + response.raise_for_status() + + data = response.json() + # API returns {data: [...], totalCount: ..., limit: ..., offset: ...} + if isinstance(data, dict) and "data" in data: + if not isinstance(data["data"], list): + raise ValueError( + f"Expected list in 'data' field, got {type(data['data']).__name__}" + ) + return data["data"] + elif isinstance(data, list): + return data + else: + raise ValueError( + f"Unexpected response format: {type(data).__name__}" + ) + + except (requests.RequestException, requests.Timeout) as e: + if attempt < self.config.max_retries: + wait_time = min( + self.config.retry_min_wait * (2**attempt), + self.config.retry_max_wait, + ) + logger.warning( + f"Request failed (attempt {attempt + 1}), retrying in {wait_time}s: {e}" + ) + time.sleep(wait_time) + else: + logger.error( + f"Request failed after {self.config.max_retries + 1} attempts: {e}" + ) + raise + # This line should never be reached but satisfies type checker + raise RuntimeError( + "Unexpected error: retry loop exited without return or raise" + ) + + def find_recent_deals_for_address( + self, + address: str, + years_back: int = 2, + radius: int = 30, + max_deals: int = 100, + deal_type: int = 2, + ) -> List[Dict[str, Any]]: + """ + Find all relevant real estate deals for a given address from the last few years. + + This is the main use case function that ties everything together. + Street deals include deals from the same building which get highest priority. + + Args: + address: The address to search for + years_back: How many years back to search (default: 2) + radius: Search radius in meters for initial coordinate search (default: 30) + Small radius since street deals cover the entire street anyway + max_deals: Maximum number of deals to return (default: 100) + deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) + + Returns: + List of deals found for the address area, with same building deals prioritized first, + then street deals, then neighborhood deals + + Raises: + ValueError: If address cannot be found or processed, or input is invalid + requests.RequestException: If API requests fail after retries + """ + # Validate inputs + address = self._validate_address(address) + years_back = self._validate_positive_int(years_back, "years_back", max_value=50) + radius = self._validate_positive_int(radius, "radius", max_value=5000) + max_deals = self._validate_positive_int(max_deals, "max_deals", max_value=10000) + validators.validate_deal_type(deal_type) + + try: + # Step 1: Get coordinates for the address + logger.info( + f"Starting search for address: {address}, dealType: {deal_type}" + ) + autocomplete_result = self.autocomplete_address(address) + + if not autocomplete_result.get("results"): + raise ValueError(f"No results found for address: {address}") + + # Get the best match (first result) + best_match = autocomplete_result["results"][0] + if "shape" not in best_match: + raise ValueError("No coordinates found in autocomplete result") + + # Parse coordinates from WKT POINT string + # Format: "POINT(longitude latitude)" + shape_str = best_match["shape"] + if not shape_str.startswith("POINT("): + raise ValueError("Invalid coordinate format in autocomplete result") + + # Extract coordinates from "POINT(x y)" + coords_str = shape_str[6:-1] # Remove "POINT(" and ")" + coords = coords_str.split() + if len(coords) != 2: + raise ValueError("Invalid coordinate format in autocomplete result") + + point = (float(coords[0]), float(coords[1])) + search_address_normalized = address.lower().strip() + logger.info(f"Found coordinates: {point}") + + # Step 2: Get deals by radius to find polygon IDs + nearby_deals = self.get_deals_by_radius(point, radius=radius) + + # Extract unique polygon IDs + polygon_ids = set() + for deal in nearby_deals: + if "polygon_id" in deal: + polygon_ids.add(str(deal["polygon_id"])) + + logger.info(f"Found {len(polygon_ids)} unique polygon IDs") + + # Step 3: Calculate date range + end_date = datetime.now() + start_date = end_date - timedelta(days=years_back * 365) + start_date_str = start_date.strftime("%Y-%m") + end_date_str = end_date.strftime("%Y-%m") + + # Step 4: Get street and neighborhood deals for each polygon + # Prioritize: same building (0) > street deals (1) > neighborhood deals (2) + building_deals = [] + street_deals = [] + neighborhood_deals = [] + seen_deals = set() # For deduplication + + for polygon_id in polygon_ids: + try: + # Get street deals first (higher priority) + current_street_deals = self.get_street_deals( + polygon_id, + limit=max_deals // 2, # Allocate more to street deals + start_date=start_date_str, + end_date=end_date_str, + deal_type=deal_type, + ) + + # Get neighborhood deals (lower priority) + current_neighborhood_deals = self.get_neighborhood_deals( + polygon_id, + limit=max_deals // 4, # Allocate less to neighborhood deals + start_date=start_date_str, + end_date=end_date_str, + deal_type=deal_type, + ) + + # Process street deals and separate building deals + for deal in current_street_deals: + # Create unique deal ID for deduplication + deal_id = f"{deal.get('dealId', '')}{deal.get('dealDate', '')}" + if deal_id not in seen_deals: + seen_deals.add(deal_id) + deal["source_polygon_id"] = polygon_id + deal["deal_source"] = "street" + + # Check if this is from the same building + # Construct address from API fields (API doesn't have single "address" field) + street = deal.get("streetNameHeb", "") + house_num = str(deal.get("houseNum", "")) + deal_address = f"{street} {house_num}".lower().strip() + if self._is_same_building( + search_address_normalized, deal_address + ): + deal["deal_source"] = "same_building" + deal["priority"] = 0 # Highest priority + building_deals.append(deal) + else: + deal["priority"] = 1 # Street deals priority + street_deals.append(deal) + + # Add neighborhood deals with lowest priority + for deal in current_neighborhood_deals: + # Create unique deal ID for deduplication + deal_id = f"{deal.get('dealId', '')}{deal.get('dealDate', '')}" + if deal_id not in seen_deals: + seen_deals.add(deal_id) + deal["source_polygon_id"] = polygon_id + deal["deal_source"] = "neighborhood" + deal["priority"] = 2 # Lowest priority + neighborhood_deals.append(deal) + + except Exception as e: + logger.warning(f"Error processing polygon {polygon_id}: {e}") + continue + + # Step 5: Combine and prioritize: building deals first, then street, then neighborhood + all_deals = building_deals + street_deals + neighborhood_deals + + # Use stable sort: first by date (newest first), then by priority + # Since Python's sort is stable, the second sort maintains date order within each priority + all_deals.sort( + key=lambda x: x.get("dealDate", "1900-01-01"), reverse=True + ) # Newest first + all_deals.sort( + key=lambda x: x.get("priority", 3) + ) # Priority first (0=building, 1=street, 2=neighborhood) + + # Limit to max_deals + if len(all_deals) > max_deals: + all_deals = all_deals[:max_deals] + + # Add price per square meter calculation and deal type info + for deal in all_deals: + price = deal.get("dealAmount", 0) + area = deal.get("assetArea", 0) + if ( + isinstance(price, (int, float)) + and isinstance(area, (int, float)) + and area > 0 + ): + deal["price_per_sqm"] = round(price / area, 2) + else: + deal["price_per_sqm"] = None + + # Add deal type description for clarity + deal["deal_type"] = deal_type + deal["deal_type_description"] = ( + "first_hand_new" if deal_type == 1 else "second_hand_used" + ) + + logger.info( + f"Found {len(all_deals)} total deals for address: {address} " + f"(Building: {len(building_deals)}, Street: {len(street_deals)}, Neighborhood: {len(neighborhood_deals)}) " + f"[{all_deals[0]['deal_type_description'] if all_deals else 'N/A'}]" + ) + return all_deals + + except Exception as e: + logger.error(f"Error in find_recent_deals_for_address: {e}") + raise + + # Filtering methods (delegate to filters module) + def filter_deals_by_criteria( + self, + deals: List[Dict[str, Any]], + property_type: Optional[str] = None, + min_rooms: Optional[float] = None, + max_rooms: Optional[float] = None, + min_price: Optional[float] = None, + max_price: Optional[float] = None, + min_area: Optional[float] = None, + max_area: Optional[float] = None, + min_floor: Optional[int] = None, + max_floor: Optional[int] = None, + ) -> List[Dict[str, Any]]: + """ + Filter deals by various criteria. + + Delegates to filters.filter_deals_by_criteria for the actual filtering logic. + """ + return filters.filter_deals_by_criteria( + deals=deals, + property_type=property_type, + min_rooms=min_rooms, + max_rooms=max_rooms, + min_price=min_price, + max_price=max_price, + min_area=min_area, + max_area=max_area, + min_floor=min_floor, + max_floor=max_floor, + ) + + # Statistics methods (delegate to statistics module) + def calculate_deal_statistics(self, deals: List[Dict[str, Any]]) -> Dict[str, Any]: + """ + Calculate statistical aggregations on deal data. + + Delegates to statistics.calculate_deal_statistics for the actual calculations. + """ + return statistics.calculate_deal_statistics(deals) + + def _calculate_std_dev(self, values: List[float]) -> float: + """ + Calculate standard deviation of a list of values. + + Delegates to statistics.calculate_std_dev for the actual calculation. + """ + return statistics.calculate_std_dev(values) + + # Market analysis methods (delegate to market_analysis module) + def _parse_deal_dates( + self, deals: List[Dict[str, Any]], time_period_months: Optional[int] = None + ): + """ + Parse and filter deal dates from a list of deals. + + Delegates to market_analysis.parse_deal_dates for the actual parsing. + """ + return market_analysis.parse_deal_dates(deals, time_period_months) + + def calculate_market_activity_score( + self, deals: List[Dict[str, Any]], time_period_months: int = 12 + ) -> Dict[str, Any]: + """ + Calculate market activity and liquidity metrics. + + Delegates to market_analysis.calculate_market_activity_score for the analysis. + """ + return market_analysis.calculate_market_activity_score( + deals, time_period_months + ) + + def analyze_investment_potential( + self, deals: List[Dict[str, Any]] + ) -> Dict[str, Any]: + """ + Analyze investment potential based on price trends and market stability. + + Delegates to market_analysis.analyze_investment_potential for the analysis. + """ + return market_analysis.analyze_investment_potential(deals) + + def get_market_liquidity( + self, deals: List[Dict[str, Any]], time_period_months: int = 12 + ) -> Dict[str, Any]: + """ + Get detailed market liquidity and turnover metrics. + + Delegates to market_analysis.get_market_liquidity for the analysis. + """ + return market_analysis.get_market_liquidity(deals, time_period_months) diff --git a/nadlan_mcp/govmap/market_analysis.py b/nadlan_mcp/govmap/market_analysis.py new file mode 100644 index 0000000..ab53fd9 --- /dev/null +++ b/nadlan_mcp/govmap/market_analysis.py @@ -0,0 +1,422 @@ +""" +Market analysis functions for real estate deal data. + +This module provides functions for analyzing market trends, activity, and investment potential. +Focused on providing data metrics; the LLM interprets them for investment advice. +""" + +import logging +from collections import defaultdict +from datetime import datetime, timedelta +from typing import Any, Dict, List, Optional, Tuple + +from .statistics import calculate_std_dev + +logger = logging.getLogger(__name__) + +# Market Activity Thresholds (deals per month) +ACTIVITY_VERY_HIGH_THRESHOLD = 10 +ACTIVITY_HIGH_THRESHOLD = 5 +ACTIVITY_MODERATE_THRESHOLD = 3 +ACTIVITY_LOW_THRESHOLD = 1 + +# Price Volatility Thresholds (coefficient of variation %) +VOLATILITY_VERY_VOLATILE_THRESHOLD = 50 +VOLATILITY_VOLATILE_THRESHOLD = 30 +VOLATILITY_MODERATE_THRESHOLD = 20 +VOLATILITY_STABLE_THRESHOLD = 10 + +# Liquidity Thresholds (deals per month) +LIQUIDITY_VERY_HIGH_THRESHOLD = 8 +LIQUIDITY_HIGH_THRESHOLD = 5 +LIQUIDITY_MODERATE_THRESHOLD = 2 +LIQUIDITY_LOW_THRESHOLD = 0.5 + + +def parse_deal_dates( + deals: List[Dict[str, Any]], time_period_months: Optional[int] = None +) -> Tuple[List[str], Dict[str, int], Dict[str, int]]: + """ + Parse and filter deal dates from a list of deals. + + This helper method centralizes the date parsing logic used across + multiple market analysis functions. It validates dates, filters by + time period if specified, and groups deals by month and quarter. + + Args: + deals: List of deal dictionaries with 'dealDate' field + time_period_months: Optional time period to filter (from today backwards) + + Returns: + Tuple containing: + - List of valid deal date strings + - Dictionary mapping year-month to deal counts + - Dictionary mapping year-quarter to deal counts + + Raises: + ValueError: If no valid deal dates are found + """ + # Calculate cutoff date if time period is specified + cutoff_date = None + if time_period_months is not None: + cutoff_date = datetime.now() - timedelta(days=time_period_months * 30) + cutoff_date_str = cutoff_date.strftime("%Y-%m-%d") + + monthly_deals = defaultdict(int) + quarterly_deals = defaultdict(int) + deal_dates = [] + + for deal in deals: + date_str = deal.get("dealDate", "") + if not date_str: + continue + + try: + # Filter by time period if specified + if cutoff_date is not None and date_str < cutoff_date_str: + continue + + # Parse date components + year = int(date_str[:4]) + month = int(date_str[5:7]) + quarter = (month - 1) // 3 + 1 # 1-4 + + # Track by month and quarter + year_month = f"{year}-{month:02d}" + year_quarter = f"{year}-Q{quarter}" + + monthly_deals[year_month] += 1 + quarterly_deals[year_quarter] += 1 + deal_dates.append(date_str) + except (ValueError, IndexError): + logger.warning(f"Invalid date format: {date_str}") + continue + + if not deal_dates: + raise ValueError("No valid deal dates found in deals list") + + return deal_dates, dict(monthly_deals), dict(quarterly_deals) + + +def calculate_market_activity_score( + deals: List[Dict[str, Any]], time_period_months: int = 12 +) -> Dict[str, Any]: + """ + Calculate market activity and liquidity metrics. + + This function analyzes deal frequency, velocity, and market activity levels + to provide a comprehensive view of market liquidity. + + Args: + deals: List of deal dictionaries + time_period_months: Time period to analyze in months (default: 12) + + Returns: + Dictionary containing: + - total_deals: Total number of deals + - deals_per_month: Average deals per month + - activity_score: Market activity score (0-100) + - trend: Activity trend ('increasing', 'stable', 'decreasing') + - monthly_distribution: Deals per month breakdown + - activity_level: Description ('very_high', 'high', 'moderate', 'low', 'very_low') + + Raises: + ValueError: If deals list is empty or invalid + """ + if not deals: + raise ValueError("Cannot calculate market activity from empty deals list") + + # Parse deal dates and group by month (with time period filtering) + deal_dates, monthly_deals, _ = parse_deal_dates(deals, time_period_months) + + # Calculate metrics + total_deals = len(deal_dates) + unique_months = len(monthly_deals) + deals_per_month = total_deals / unique_months if unique_months > 0 else 0 + + # Calculate activity score (0-100) + # Based on deals per month using defined thresholds + if deals_per_month >= ACTIVITY_VERY_HIGH_THRESHOLD: + activity_score = 100 + activity_level = "very_high" + elif deals_per_month >= ACTIVITY_HIGH_THRESHOLD: + activity_score = 75 + ((deals_per_month - ACTIVITY_HIGH_THRESHOLD) / ACTIVITY_HIGH_THRESHOLD) * 25 + activity_level = "high" + elif deals_per_month >= ACTIVITY_MODERATE_THRESHOLD: + activity_score = 50 + ((deals_per_month - ACTIVITY_MODERATE_THRESHOLD) / (ACTIVITY_HIGH_THRESHOLD - ACTIVITY_MODERATE_THRESHOLD)) * 25 + activity_level = "moderate" + elif deals_per_month >= ACTIVITY_LOW_THRESHOLD: + activity_score = 25 + ((deals_per_month - ACTIVITY_LOW_THRESHOLD) / (ACTIVITY_MODERATE_THRESHOLD - ACTIVITY_LOW_THRESHOLD)) * 25 + activity_level = "low" + else: + activity_score = deals_per_month * 25 + activity_level = "very_low" + + # Calculate trend (compare first half vs second half) + sorted_months = sorted(monthly_deals.keys()) + if len(sorted_months) >= 4: + mid_point = len(sorted_months) // 2 + first_half_avg = sum(monthly_deals[m] for m in sorted_months[:mid_point]) / mid_point + second_half_avg = sum(monthly_deals[m] for m in sorted_months[mid_point:]) / ( + len(sorted_months) - mid_point + ) + + change_ratio = (second_half_avg - first_half_avg) / first_half_avg if first_half_avg > 0 else 0 + + if change_ratio > 0.15: + trend = "increasing" + elif change_ratio < -0.15: + trend = "decreasing" + else: + trend = "stable" + else: + trend = "insufficient_data" + + return { + "total_deals": total_deals, + "unique_months": unique_months, + "deals_per_month": round(deals_per_month, 2), + "activity_score": round(activity_score, 1), + "activity_level": activity_level, + "trend": trend, + "monthly_distribution": dict(sorted(monthly_deals.items())), + } + + +def analyze_investment_potential(deals: List[Dict[str, Any]]) -> Dict[str, Any]: + """ + Analyze investment potential based on price trends and market stability. + + This function calculates price appreciation rates, market volatility, + and provides investment metrics for decision-making. The MCP provides + data metrics; the LLM interprets them for investment advice. + + Args: + deals: List of deal dictionaries with price and date information + + Returns: + Dictionary containing: + - price_appreciation_rate: Annual price growth rate (%) + - price_volatility: Price volatility score (0-100, lower is more stable) + - market_stability: Stability rating ('very_stable', 'stable', 'moderate', 'volatile', 'very_volatile') + - price_trend: Price direction ('increasing', 'stable', 'decreasing') + - avg_price_per_sqm: Average price per square meter + - price_change_pct: Total price change percentage + - investment_score: Overall investment score (0-100) + - data_quality: Quality of data ('excellent', 'good', 'fair', 'limited') + + Raises: + ValueError: If deals list is empty or lacks required data + """ + if not deals: + raise ValueError("Cannot analyze investment potential from empty deals list") + + # Extract price per sqm and dates + price_data = [] + for deal in deals: + price_per_sqm = deal.get("price_per_sqm") + date_str = deal.get("dealDate", "") + + if isinstance(price_per_sqm, (int, float)) and price_per_sqm > 0 and date_str: + try: + # Parse date for sorting + year = int(date_str[:4]) + month = int(date_str[5:7]) + price_data.append((year + month / 12.0, price_per_sqm)) + except (ValueError, IndexError): + continue + + if len(price_data) < 3: + raise ValueError( + "Insufficient data for investment analysis (need at least 3 valid deals with price and date)" + ) + + # Sort by time + price_data.sort(key=lambda x: x[0]) + times = [p[0] for p in price_data] + prices = [p[1] for p in price_data] + + # Calculate average price + avg_price_per_sqm = sum(prices) / len(prices) + + # Calculate price appreciation rate (using linear regression approximation) + n = len(price_data) + sum_t = sum(times) + sum_p = sum(prices) + sum_tp = sum(t * p for t, p in price_data) + sum_t2 = sum(t * t for t in times) + + # Linear regression slope + if n * sum_t2 - sum_t * sum_t != 0: + slope = (n * sum_tp - sum_t * sum_p) / (n * sum_t2 - sum_t * sum_t) + # Convert to annual percentage change + price_appreciation_rate = (slope / avg_price_per_sqm) * 100 if avg_price_per_sqm > 0 else 0 + else: + price_appreciation_rate = 0 + + # Calculate price change from first to last deal + if prices[0] > 0: + price_change_pct = ((prices[-1] - prices[0]) / prices[0]) * 100 + else: + price_change_pct = 0 + + # Determine price trend + if price_appreciation_rate > 2: + price_trend = "increasing" + elif price_appreciation_rate < -2: + price_trend = "decreasing" + 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 + else: + coefficient_of_variation = 0 + + # Convert CV to volatility score (0-100, lower is better) + # Using defined volatility thresholds + if coefficient_of_variation > VOLATILITY_VERY_VOLATILE_THRESHOLD: + volatility_score = 100 + market_stability = "very_volatile" + elif coefficient_of_variation > VOLATILITY_VOLATILE_THRESHOLD: + volatility_score = 75 + ((coefficient_of_variation - VOLATILITY_VOLATILE_THRESHOLD) / (VOLATILITY_VERY_VOLATILE_THRESHOLD - VOLATILITY_VOLATILE_THRESHOLD)) * 25 + market_stability = "volatile" + elif coefficient_of_variation > VOLATILITY_MODERATE_THRESHOLD: + volatility_score = 50 + ((coefficient_of_variation - VOLATILITY_MODERATE_THRESHOLD) / (VOLATILITY_VOLATILE_THRESHOLD - VOLATILITY_MODERATE_THRESHOLD)) * 25 + market_stability = "moderate" + elif coefficient_of_variation > VOLATILITY_STABLE_THRESHOLD: + volatility_score = 25 + ((coefficient_of_variation - VOLATILITY_STABLE_THRESHOLD) / (VOLATILITY_MODERATE_THRESHOLD - VOLATILITY_STABLE_THRESHOLD)) * 25 + market_stability = "stable" + else: + volatility_score = (coefficient_of_variation / VOLATILITY_STABLE_THRESHOLD) * 25 + market_stability = "very_stable" + + # Calculate investment score (0-100) + # Positive: price appreciation, market stability (low volatility) + # Negative: price decline, high volatility + appreciation_component = min(max(price_appreciation_rate * 5, -25), 50) # -25 to +50 + stability_component = (100 - volatility_score) * 0.5 # 0 to 50 + + investment_score = max(0, min(100, appreciation_component + stability_component)) + + # Data quality assessment + if n >= 20: + data_quality = "excellent" + elif n >= 10: + data_quality = "good" + elif n >= 5: + data_quality = "fair" + else: + data_quality = "limited" + + return { + "price_appreciation_rate": round(price_appreciation_rate, 2), + "price_volatility": round(volatility_score, 1), + "market_stability": market_stability, + "price_trend": price_trend, + "avg_price_per_sqm": round(avg_price_per_sqm, 0), + "price_change_pct": round(price_change_pct, 2), + "investment_score": round(investment_score, 1), + "data_quality": data_quality, + "sample_size": n, + } + + +def get_market_liquidity( + deals: List[Dict[str, Any]], time_period_months: int = 12 +) -> Dict[str, Any]: + """ + Get detailed market liquidity and turnover metrics. + + This function provides granular liquidity metrics including deal velocity, + quarterly trends, and market turnover indicators. + + Args: + deals: List of deal dictionaries + time_period_months: Time period to analyze in months (default: 12) + + Returns: + Dictionary containing: + - total_deals: Total number of deals in period + - deals_per_month: Average deals per month + - deals_per_quarter: Average deals per quarter + - quarterly_breakdown: Deals grouped by quarter + - velocity_score: Market velocity score (0-100) + - liquidity_rating: Liquidity rating ('very_high', 'high', 'moderate', 'low', 'very_low') + - trend_direction: Trend in liquidity ('improving', 'stable', 'declining') + - most_active_period: Quarter/month with most activity + + Raises: + ValueError: If deals list is empty or invalid + """ + if not deals: + raise ValueError("Cannot calculate market liquidity from empty deals list") + + # Parse deal dates and group by month and quarter (with time period filtering) + deal_dates, monthly_deals, quarterly_deals = parse_deal_dates(deals, time_period_months) + + # Calculate metrics + total_deals = len(deal_dates) + unique_months = len(monthly_deals) + unique_quarters = len(quarterly_deals) + + deals_per_month = total_deals / unique_months if unique_months > 0 else 0 + deals_per_quarter = total_deals / unique_quarters if unique_quarters > 0 else 0 + + # Calculate velocity score (similar to activity score but focused on turnover) + # Based on monthly deal velocity using defined thresholds + if deals_per_month >= LIQUIDITY_VERY_HIGH_THRESHOLD: + velocity_score = 100 + liquidity_rating = "very_high" + elif deals_per_month >= LIQUIDITY_HIGH_THRESHOLD: + velocity_score = 75 + ((deals_per_month - LIQUIDITY_HIGH_THRESHOLD) / (LIQUIDITY_VERY_HIGH_THRESHOLD - LIQUIDITY_HIGH_THRESHOLD)) * 25 + liquidity_rating = "high" + elif deals_per_month >= LIQUIDITY_MODERATE_THRESHOLD: + velocity_score = 50 + ((deals_per_month - LIQUIDITY_MODERATE_THRESHOLD) / (LIQUIDITY_HIGH_THRESHOLD - LIQUIDITY_MODERATE_THRESHOLD)) * 25 + liquidity_rating = "moderate" + elif deals_per_month >= LIQUIDITY_LOW_THRESHOLD: + velocity_score = 25 + ((deals_per_month - LIQUIDITY_LOW_THRESHOLD) / (LIQUIDITY_MODERATE_THRESHOLD - LIQUIDITY_LOW_THRESHOLD)) * 25 + liquidity_rating = "low" + else: + velocity_score = deals_per_month * 50 + liquidity_rating = "very_low" + + # Determine trend direction (compare recent quarter to earlier quarters) + sorted_quarters = sorted(quarterly_deals.keys()) + if len(sorted_quarters) >= 3: + recent_quarter_avg = quarterly_deals[sorted_quarters[-1]] + earlier_quarters_avg = sum(quarterly_deals[q] for q in sorted_quarters[:-1]) / ( + len(sorted_quarters) - 1 + ) + + if recent_quarter_avg > earlier_quarters_avg * 1.2: + trend_direction = "improving" + elif recent_quarter_avg < earlier_quarters_avg * 0.8: + trend_direction = "declining" + else: + trend_direction = "stable" + else: + trend_direction = "insufficient_data" + + # Find most active period + if quarterly_deals: + most_active_quarter = max(quarterly_deals.items(), key=lambda x: x[1]) + most_active_period = f"{most_active_quarter[0]} ({most_active_quarter[1]} deals)" + else: + most_active_period = "N/A" + + return { + "total_deals": total_deals, + "unique_months": unique_months, + "unique_quarters": unique_quarters, + "deals_per_month": round(deals_per_month, 2), + "deals_per_quarter": round(deals_per_quarter, 2), + "quarterly_breakdown": dict(sorted(quarterly_deals.items())), + "monthly_breakdown": dict(sorted(monthly_deals.items())), + "velocity_score": round(velocity_score, 1), + "liquidity_rating": liquidity_rating, + "trend_direction": trend_direction, + "most_active_period": most_active_period, + }