mcp-servers/nadlan-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implementation of phase 4.1

Browse Source
This commit is contained in:
Nitzan Pomerantz
2025-10-26 10:58:46 +02:00
parent 4cb70c547b
commit 0ac9d136cd
13 changed files with 2115 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nadlan_mcp/govmap/__init__.py
+27 -1
View File
@@ -5,7 +5,8 @@ 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)
- GovmapClient: Main API client class
- Pydantic models for type-safe data structures
- filter_deals_by_criteria: Filter deals by various criteria
- calculate_deal_statistics: Calculate statistical aggregations
- calculate_market_activity_score: Market activity and trend metrics
@@ -13,6 +14,20 @@ Public API:
- get_market_liquidity: Market liquidity and velocity metrics
"""
# Pydantic models
from .models import (
CoordinatePoint,
Address,
AutocompleteResult,
AutocompleteResponse,
Deal,
DealStatistics,
MarketActivityScore,
InvestmentAnalysis,
LiquidityMetrics,
DealFilters,
)
# Filter functions
from .filters import filter_deals_by_criteria
@@ -44,6 +59,17 @@ from .client import GovmapClient
__all__ = [
# Main client class
"GovmapClient",
# Pydantic models
"CoordinatePoint",
"Address",
"AutocompleteResult",
"AutocompleteResponse",
"Deal",
"DealStatistics",
"MarketActivityScore",
"InvestmentAnalysis",
"LiquidityMetrics",
"DealFilters",
# Filtering
"filter_deals_by_criteria",
# Statistics
nadlan_mcp/govmap/client.py
+186 -77
View File
@@ -15,6 +15,18 @@ import requests
from nadlan_mcp.config import GovmapConfig, get_config
# Import models
from .models import (
Deal,
AutocompleteResponse,
AutocompleteResult,
CoordinatePoint,
DealStatistics,
MarketActivityScore,
InvestmentAnalysis,
LiquidityMetrics,
)
# Import functions from modular package
from . import validators
from . import utils
@@ -96,7 +108,7 @@ class GovmapClient:
return utils.extract_floor_number(floor_str)
# Core API methods
def autocomplete_address(self, search_text: str) -> Dict[str, Any]:
def autocomplete_address(self, search_text: str) -> AutocompleteResponse:
"""
Find the most likely match for a given address using autocomplete.
@@ -104,7 +116,7 @@ class GovmapClient:
search_text: The address to search for (e.g., "סוקולוב 38 חולון")
Returns:
Dict containing the JSON response from the API with coordinates
AutocompleteResponse model with results and coordinates
Raises:
requests.RequestException: If the API request fails after retries
@@ -136,7 +148,37 @@ class GovmapClient:
if not data or "results" not in data:
raise ValueError("Invalid response format from autocomplete API")
return data
# Parse results into AutocompleteResult models
results = []
for result in data.get("results", []):
# Parse coordinates from WKT POINT format if available
coordinates = None
shape_str = result.get("shape", "")
if shape_str and shape_str.startswith("POINT("):
try:
coords_str = shape_str[6:-1] # Remove "POINT(" and ")"
coords = coords_str.split()
if len(coords) == 2:
coordinates = CoordinatePoint(
longitude=float(coords[0]),
latitude=float(coords[1])
)
except (ValueError, IndexError) as e:
logger.warning(f"Failed to parse coordinates from shape: {shape_str}, error: {e}")
results.append(AutocompleteResult(
text=result.get("text", ""),
id=result.get("id", ""),
type=result.get("type", ""),
score=result.get("score", 0),
coordinates=coordinates,
shape=shape_str if shape_str else None,
))
return AutocompleteResponse(
resultsCount=data.get("resultsCount", len(results)),
results=results
)
except (requests.RequestException, requests.Timeout) as e:
if attempt < self.config.max_retries:
@@ -214,7 +256,7 @@ class GovmapClient:
def get_deals_by_radius(
self, point: Tuple[float, float], radius: int = 50
) -> List[Dict[str, Any]]:
) -> List[Deal]:
"""
Find real estate deals within a specified radius of a point.
@@ -223,7 +265,7 @@ class GovmapClient:
radius: The search radius in meters (default: 50)
Returns:
List of deals found within the radius
List of Deal models found within the radius
Raises:
requests.RequestException: If the API request fails after retries
@@ -250,7 +292,18 @@ class GovmapClient:
raise ValueError(
f"Expected list response, got {type(data).__name__}"
)
return data
# Parse each deal dict into Deal model
deals = []
for deal_dict in data:
try:
deal = Deal.model_validate(deal_dict)
deals.append(deal)
except Exception as e:
logger.warning(f"Failed to parse deal: {e}. Skipping deal.")
continue
return deals
except (requests.RequestException, requests.Timeout) as e:
if attempt < self.config.max_retries:
@@ -279,7 +332,7 @@ class GovmapClient:
start_date: Optional[str] = None,
end_date: Optional[str] = None,
deal_type: int = 2,
) -> List[Dict[str, Any]]:
) -> List[Deal]:
"""
Retrieve detailed information about deals on a specific street.
@@ -291,7 +344,7 @@ class GovmapClient:
deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2)
Returns:
List of detailed deal information for the street
List of Deal models for the street
Raises:
requests.RequestException: If the API request fails after retries
@@ -308,7 +361,7 @@ class GovmapClient:
url = f"{self.base_url}/real-estate/street-deals/{polygon_id}"
params: Dict[str, Any] = {"limit": limit, "dealType": deal_type}
params = {"limit": limit, "dealType": deal_type}
if start_date:
params["startDate"] = start_date
if end_date:
@@ -328,19 +381,32 @@ class GovmapClient:
data = response.json()
# API returns {data: [...], totalCount: ..., limit: ..., offset: ...}
deal_dicts = []
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"]
deal_dicts = data["data"]
elif isinstance(data, list):
return data
deal_dicts = data
else:
raise ValueError(
f"Unexpected response format: {type(data).__name__}"
)
# Parse each deal dict into Deal model
deals = []
for deal_dict in deal_dicts:
try:
deal = Deal.model_validate(deal_dict)
deals.append(deal)
except Exception as e:
logger.warning(f"Failed to parse deal: {e}. Skipping deal.")
continue
return deals
except (requests.RequestException, requests.Timeout) as e:
if attempt < self.config.max_retries:
wait_time = min(
@@ -368,7 +434,7 @@ class GovmapClient:
start_date: Optional[str] = None,
end_date: Optional[str] = None,
deal_type: int = 2,
) -> List[Dict[str, Any]]:
) -> List[Deal]:
"""
Retrieve deals within the same neighborhood as the given polygon_id.
@@ -380,7 +446,7 @@ class GovmapClient:
deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2)
Returns:
List of deals in the neighborhood
List of Deal models in the neighborhood
Raises:
requests.RequestException: If the API request fails after retries
@@ -397,7 +463,7 @@ class GovmapClient:
url = f"{self.base_url}/real-estate/neighborhood-deals/{polygon_id}"
params: Dict[str, Any] = {"limit": limit, "dealType": deal_type}
params = {"limit": limit, "dealType": deal_type}
if start_date:
params["startDate"] = start_date
if end_date:
@@ -417,19 +483,32 @@ class GovmapClient:
data = response.json()
# API returns {data: [...], totalCount: ..., limit: ..., offset: ...}
deal_dicts = []
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"]
deal_dicts = data["data"]
elif isinstance(data, list):
return data
deal_dicts = data
else:
raise ValueError(
f"Unexpected response format: {type(data).__name__}"
)
# Parse each deal dict into Deal model
deals = []
for deal_dict in deal_dicts:
try:
deal = Deal.model_validate(deal_dict)
deals.append(deal)
except Exception as e:
logger.warning(f"Failed to parse deal: {e}. Skipping deal.")
continue
return deals
except (requests.RequestException, requests.Timeout) as e:
if attempt < self.config.max_retries:
wait_time = min(
@@ -457,7 +536,7 @@ class GovmapClient:
radius: int = 30,
max_deals: int = 100,
deal_type: int = 2,
) -> List[Dict[str, Any]]:
) -> List[Deal]:
"""
Find all relevant real estate deals for a given address from the last few years.
@@ -473,7 +552,7 @@ class GovmapClient:
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,
List of Deal models found for the address area, with same building deals prioritized first,
then street deals, then neighborhood deals
Raises:
@@ -494,27 +573,16 @@ class GovmapClient:
)
autocomplete_result = self.autocomplete_address(address)
if not autocomplete_result.get("results"):
if not autocomplete_result.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:
best_match = autocomplete_result.results[0]
if not best_match.coordinates:
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]))
# Use coordinates from the model
point = (best_match.coordinates.longitude, best_match.coordinates.latitude)
search_address_normalized = address.lower().strip()
logger.info(f"Found coordinates: {point}")
@@ -524,8 +592,10 @@ class GovmapClient:
# Extract unique polygon IDs
polygon_ids = set()
for deal in nearby_deals:
if "polygon_id" in deal:
polygon_ids.add(str(deal["polygon_id"]))
# Try to get polygon_id from the deal model
polygon_id = getattr(deal, 'polygon_id', None) or deal.source_polygon_id
if polygon_id:
polygon_ids.add(str(polygon_id))
logger.info(f"Found {len(polygon_ids)} unique polygon IDs")
@@ -565,36 +635,38 @@ class GovmapClient:
# 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', '')}"
deal_id = f"{deal.objectid}{deal.deal_date}"
if deal_id not in seen_deals:
seen_deals.add(deal_id)
deal["source_polygon_id"] = polygon_id
deal["deal_source"] = "street"
# Store metadata using dynamic attributes (allowed by extra='allow')
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", ""))
# Construct address from model fields
street = deal.street_name or ""
house_num = str(deal.house_number or "")
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
deal.deal_source = "same_building"
deal.priority = 0 # Highest priority
building_deals.append(deal)
else:
deal["priority"] = 1 # Street deals priority
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', '')}"
deal_id = f"{deal.objectid}{deal.deal_date}"
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
# Store metadata using dynamic attributes
deal.source_polygon_id = polygon_id
deal.deal_source = "neighborhood"
deal.priority = 2 # Lowest priority
neighborhood_deals.append(deal)
except Exception as e:
@@ -607,39 +679,28 @@ class GovmapClient:
# 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
key=lambda x: x.deal_date or "1900-01-01", reverse=True
) # Newest first
all_deals.sort(
key=lambda x: x.get("priority", 3)
key=lambda x: getattr(x, '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
# Add deal type metadata for clarity
# Note: price_per_sqm is now a computed field on the Deal model
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"] = (
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'}]"
f"[{all_deals[0].deal_type_description if all_deals else 'N/A'}]"
)
return all_deals
@@ -650,7 +711,7 @@ class GovmapClient:
# Filtering methods (delegate to filters module)
def filter_deals_by_criteria(
self,
deals: List[Dict[str, Any]],
deals: List[Deal],
property_type: Optional[str] = None,
min_rooms: Optional[float] = None,
max_rooms: Optional[float] = None,
@@ -660,11 +721,26 @@ class GovmapClient:
max_area: Optional[float] = None,
min_floor: Optional[int] = None,
max_floor: Optional[int] = None,
) -> List[Dict[str, Any]]:
) -> List[Deal]:
"""
Filter deals by various criteria.
Delegates to filters.filter_deals_by_criteria for the actual filtering logic.
Args:
deals: List of Deal model instances to filter
property_type: Property type to filter by (Hebrew description)
min_rooms: Minimum number of rooms
max_rooms: Maximum number of rooms
min_price: Minimum deal amount
max_price: Maximum deal amount
min_area: Minimum asset area (square meters)
max_area: Maximum asset area (square meters)
min_floor: Minimum floor number
max_floor: Maximum floor number
Returns:
Filtered list of Deal instances
"""
return filters.filter_deals_by_criteria(
deals=deals,
@@ -680,11 +756,17 @@ class GovmapClient:
)
# Statistics methods (delegate to statistics module)
def calculate_deal_statistics(self, deals: List[Dict[str, Any]]) -> Dict[str, Any]:
def calculate_deal_statistics(self, deals: List[Deal]) -> DealStatistics:
"""
Calculate statistical aggregations on deal data.
Delegates to statistics.calculate_deal_statistics for the actual calculations.
Args:
deals: List of Deal model instances
Returns:
DealStatistics model with comprehensive metrics
"""
return statistics.calculate_deal_statistics(deals)
@@ -698,43 +780,70 @@ class GovmapClient:
# Market analysis methods (delegate to market_analysis module)
def _parse_deal_dates(
self, deals: List[Dict[str, Any]], time_period_months: Optional[int] = None
self, deals: List[Deal], 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.
Args:
deals: List of Deal model instances
time_period_months: Optional time period to filter (from today backwards)
Returns:
Tuple containing deal dates, monthly distribution, and quarterly distribution
"""
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]:
self, deals: List[Deal], time_period_months: int = 12
) -> MarketActivityScore:
"""
Calculate market activity and liquidity metrics.
Delegates to market_analysis.calculate_market_activity_score for the analysis.
Args:
deals: List of Deal model instances
time_period_months: Time period to analyze in months (default: 12)
Returns:
MarketActivityScore model with activity metrics
"""
return market_analysis.calculate_market_activity_score(
deals, time_period_months
)
def analyze_investment_potential(
self, deals: List[Dict[str, Any]]
) -> Dict[str, Any]:
self, deals: List[Deal]
) -> InvestmentAnalysis:
"""
Analyze investment potential based on price trends and market stability.
Delegates to market_analysis.analyze_investment_potential for the analysis.
Args:
deals: List of Deal model instances with price and date information
Returns:
InvestmentAnalysis model with investment metrics
"""
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]:
self, deals: List[Deal], time_period_months: int = 12
) -> LiquidityMetrics:
"""
Get detailed market liquidity and turnover metrics.
Delegates to market_analysis.get_market_liquidity for the analysis.
Args:
deals: List of Deal model instances
time_period_months: Time period to analyze in months (default: 12)
Returns:
LiquidityMetrics model with liquidity and velocity metrics
"""
return market_analysis.get_market_liquidity(deals, time_period_months)
nadlan_mcp/govmap/filters.py
+54 -44
View File
@@ -4,13 +4,15 @@ Deal filtering functions.
This module provides composable functions for filtering real estate deal data.
"""
from typing import Any, Dict, List, Optional
from typing import List, Optional, Union
from .models import Deal, DealFilters
from .utils import extract_floor_number
def filter_deals_by_criteria(
deals: List[Dict[str, Any]],
deals: List[Deal],
filters: Optional[Union[DealFilters, dict]] = None,
property_type: Optional[str] = None,
min_rooms: Optional[float] = None,
max_rooms: Optional[float] = None,
@@ -20,12 +22,16 @@ def filter_deals_by_criteria(
max_area: Optional[float] = None,
min_floor: Optional[int] = None,
max_floor: Optional[int] = None,
) -> List[Dict[str, Any]]:
) -> List[Deal]:
"""
Filter deals by various criteria.
Can accept either a DealFilters model or individual filter parameters.
Individual parameters take precedence over filters model.
Args:
deals: List of deal dictionaries to filter
deals: List of Deal model instances to filter
filters: Optional DealFilters model or dict with filter criteria
property_type: Property type to filter by (Hebrew description)
min_rooms: Minimum number of rooms
max_rooms: Maximum number of rooms
@@ -37,7 +43,7 @@ def filter_deals_by_criteria(
max_floor: Maximum floor number
Returns:
Filtered list of deals
Filtered list of Deal instances
Raises:
ValueError: If filter criteria are invalid
@@ -45,7 +51,23 @@ def filter_deals_by_criteria(
if not isinstance(deals, list):
raise ValueError("deals must be a list")
# Validate numeric ranges
# Convert filters dict to DealFilters model if needed
if isinstance(filters, dict):
filters = DealFilters(**filters)
# If filters model provided, use its values as defaults
if filters:
property_type = property_type or filters.property_type
min_rooms = min_rooms if min_rooms is not None else filters.min_rooms
max_rooms = max_rooms if max_rooms is not None else filters.max_rooms
min_price = min_price if min_price is not None else filters.min_price
max_price = max_price if max_price is not None else filters.max_price
min_area = min_area if min_area is not None else filters.min_area
max_area = max_area if max_area is not None else filters.max_area
min_floor = min_floor if min_floor is not None else filters.min_floor
max_floor = max_floor if max_floor is not None else filters.max_floor
# Validate numeric ranges (Pydantic validates these too, but check anyway)
if min_rooms is not None and max_rooms is not None and min_rooms > max_rooms:
raise ValueError("min_rooms cannot be greater than max_rooms")
if min_price is not None and max_price is not None and min_price > max_price:
@@ -60,9 +82,7 @@ def filter_deals_by_criteria(
for deal in deals:
# Property type filter
if property_type is not None:
deal_type = deal.get(
"propertyTypeDescription", deal.get("assetTypeHeb", "")
)
deal_type = deal.property_type_description
# Skip deals with missing property type data when filter is active
if not deal_type:
continue
@@ -78,57 +98,47 @@ def filter_deals_by_criteria(
# Room count filter
if min_rooms is not None or max_rooms is not None:
rooms = deal.get("assetRoomNum")
rooms = deal.rooms
if rooms is None:
continue # Skip deals with missing room data when filter is active
try:
rooms = float(rooms)
if min_rooms is not None and rooms < min_rooms:
continue
if max_rooms is not None and rooms > max_rooms:
continue
except (TypeError, ValueError):
continue # Skip deals with invalid room data when filter is active
if min_rooms is not None and rooms < min_rooms:
continue
if max_rooms is not None and rooms > max_rooms:
continue
# Price filter
if min_price is not None or max_price is not None:
price = deal.get("dealAmount")
price = deal.deal_amount
if price is None:
continue # Skip deals with missing price data when filter is active
try:
price = float(price)
if min_price is not None and price < min_price:
continue
if max_price is not None and price > max_price:
continue
except (TypeError, ValueError):
continue # Skip deals with invalid price data when filter is active
if min_price is not None and price < min_price:
continue
if max_price is not None and price > max_price:
continue
# Area filter
if min_area is not None or max_area is not None:
area = deal.get("assetArea")
area = deal.asset_area
if area is None:
continue # Skip deals with missing area data when filter is active
try:
area = float(area)
if min_area is not None and area < min_area:
continue
if max_area is not None and area > max_area:
continue
except (TypeError, ValueError):
continue # Skip deals with invalid area data when filter is active
if min_area is not None and area < min_area:
continue
if max_area is not None and area > max_area:
continue
# Floor filter
if min_floor is not None or max_floor is not None:
floor_str = deal.get("floorNo", "")
if floor_str and isinstance(floor_str, str):
# Use floor_number if available, otherwise try to parse floor description
floor_num = deal.floor_number
if floor_num is None and deal.floor:
# Try to extract floor number (handles Hebrew floor descriptions)
floor_num = extract_floor_number(floor_str)
if floor_num is not None:
if min_floor is not None and floor_num < min_floor:
continue
if max_floor is not None and floor_num > max_floor:
continue
floor_num = extract_floor_number(deal.floor)
if floor_num is not None:
if min_floor is not None and floor_num < min_floor:
continue
if max_floor is not None and floor_num > max_floor:
continue
filtered_deals.append(deal)
nadlan_mcp/govmap/market_analysis.py
+44 -62
View File
@@ -8,8 +8,9 @@ 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 typing import Dict, List, Optional, Tuple
from .models import Deal, MarketActivityScore, InvestmentAnalysis, LiquidityMetrics
from .statistics import calculate_std_dev
logger = logging.getLogger(__name__)
@@ -34,7 +35,7 @@ LIQUIDITY_LOW_THRESHOLD = 0.5
def parse_deal_dates(
deals: List[Dict[str, Any]], time_period_months: Optional[int] = None
deals: List[Deal], 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.
@@ -44,7 +45,7 @@ def parse_deal_dates(
time period if specified, and groups deals by month and quarter.
Args:
deals: List of deal dictionaries with 'dealDate' field
deals: List of Deal model instances
time_period_months: Optional time period to filter (from today backwards)
Returns:
@@ -67,7 +68,7 @@ def parse_deal_dates(
deal_dates = []
for deal in deals:
date_str = deal.get("dealDate", "")
date_str = deal.deal_date
if not date_str:
continue
@@ -99,8 +100,8 @@ def parse_deal_dates(
def calculate_market_activity_score(
deals: List[Dict[str, Any]], time_period_months: int = 12
) -> Dict[str, Any]:
deals: List[Deal], time_period_months: int = 12
) -> MarketActivityScore:
"""
Calculate market activity and liquidity metrics.
@@ -108,17 +109,16 @@ def calculate_market_activity_score(
to provide a comprehensive view of market liquidity.
Args:
deals: List of deal dictionaries
deals: List of Deal model instances
time_period_months: Time period to analyze in months (default: 12)
Returns:
Dictionary containing:
MarketActivityScore model with:
- 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
@@ -138,19 +138,14 @@ def calculate_market_activity_score(
# 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())
@@ -172,18 +167,17 @@ def calculate_market_activity_score(
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())),
}
return MarketActivityScore(
activity_score=round(activity_score, 1),
total_deals=total_deals,
deals_per_month=round(deals_per_month, 2),
trend=trend,
time_period_months=time_period_months,
monthly_distribution=dict(sorted(monthly_deals.items())),
)
def analyze_investment_potential(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
def analyze_investment_potential(deals: List[Deal]) -> InvestmentAnalysis:
"""
Analyze investment potential based on price trends and market stability.
@@ -192,10 +186,10 @@ def analyze_investment_potential(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
data metrics; the LLM interprets them for investment advice.
Args:
deals: List of deal dictionaries with price and date information
deals: List of Deal model instances with price and date information
Returns:
Dictionary containing:
InvestmentAnalysis model 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')
@@ -214,10 +208,10 @@ def analyze_investment_potential(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
# 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", "")
price_per_sqm = deal.price_per_sqm # Use computed field from Deal model
date_str = deal.deal_date
if isinstance(price_per_sqm, (int, float)) and price_per_sqm > 0 and date_str:
if price_per_sqm and price_per_sqm > 0 and date_str:
try:
# Parse date for sorting
year = int(date_str[:4])
@@ -311,22 +305,22 @@ def analyze_investment_potential(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
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,
}
return InvestmentAnalysis(
investment_score=round(investment_score, 1),
price_trend=price_trend,
price_appreciation_rate=round(price_appreciation_rate, 2),
price_volatility=round(volatility_score, 1),
market_stability=market_stability,
avg_price_per_sqm=round(avg_price_per_sqm, 0),
price_change_pct=round(price_change_pct, 2),
total_deals=n,
data_quality=data_quality,
)
def get_market_liquidity(
deals: List[Dict[str, Any]], time_period_months: int = 12
) -> Dict[str, Any]:
deals: List[Deal], time_period_months: int = 12
) -> LiquidityMetrics:
"""
Get detailed market liquidity and turnover metrics.
@@ -400,23 +394,11 @@ def get_market_liquidity(
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,
}
return LiquidityMetrics(
liquidity_score=round(velocity_score, 1),
total_deals=total_deals,
time_period_months=time_period_months,
avg_deals_per_month=round(deals_per_month, 2),
deal_velocity=round(deals_per_month, 2),
market_activity_level=liquidity_rating,
)
nadlan_mcp/govmap/models.py
+337
View File
@@ -0,0 +1,337 @@
"""
Pydantic models for Govmap API data structures.
This module defines type-safe models for all data structures used in the
Israeli real estate MCP system. Models provide validation, serialization,
and type safety throughout the codebase.
"""
from datetime import datetime
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field, field_validator, computed_field, ConfigDict
class CoordinatePoint(BaseModel):
"""
ITM (Israeli Transverse Mercator) coordinate point.
Attributes:
longitude: X coordinate in ITM projection (meters)
latitude: Y coordinate in ITM projection (meters)
"""
longitude: float = Field(..., description="X coordinate in ITM projection (meters)")
latitude: float = Field(..., description="Y coordinate in ITM projection (meters)")
model_config = ConfigDict(frozen=True) # Immutable coordinates
class Address(BaseModel):
"""
Israeli address with coordinates and metadata.
Attributes:
text: Full address text (Hebrew or English)
id: Unique identifier for the address
type: Address type (e.g., 'address', 'street', 'city')
score: Relevance score from autocomplete
coordinates: ITM coordinate point
"""
text: str = Field(..., description="Full address text")
id: str = Field(..., description="Unique address identifier")
type: str = Field(..., description="Address type")
score: float = Field(default=0, description="Relevance score")
coordinates: Optional[CoordinatePoint] = Field(default=None, description="ITM coordinates")
class AutocompleteResult(BaseModel):
"""
Single result from address autocomplete API.
Attributes:
text: Display text for the address
id: Unique identifier
type: Result type (address, street, city, etc.)
score: Relevance score
coordinates: Optional coordinate point
shape: Original WKT shape string from API
"""
text: str
id: str
type: str
score: float = 0
coordinates: Optional[CoordinatePoint] = None
shape: Optional[str] = None # Original WKT POINT string from API
class AutocompleteResponse(BaseModel):
"""
Response from address autocomplete API.
Attributes:
results_count: Number of results returned
results: List of autocomplete results
"""
results_count: int = Field(alias="resultsCount")
results: List[AutocompleteResult] = Field(default_factory=list)
model_config = ConfigDict(populate_by_name=True)
class Deal(BaseModel):
"""
Real estate deal from Govmap API.
Represents a single property transaction with all available details.
Most fields are optional as the API doesn't guarantee all data.
Attributes:
objectid: Unique deal identifier
deal_amount: Transaction amount in NIS
deal_date: Date of transaction
asset_area: Property area in square meters
settlement_name_heb: City/settlement name in Hebrew
property_type_description: Type of property (דירה, בית, etc.)
neighborhood: Neighborhood name
street_name: Street name
house_number: House number
floor: Floor description (may be Hebrew text)
floor_number: Parsed numeric floor number
rooms: Number of rooms
priority: Priority level for sorting (0=same building, 1=street, 2=neighborhood)
shape: WKT geometry (usually MULTIPOLYGON)
source_polygon_id: Source polygon ID
sourceorder: Source ordering
"""
# Required fields
objectid: int = Field(..., description="Unique deal identifier")
deal_amount: float = Field(..., alias="dealAmount", description="Transaction amount in NIS")
deal_date: str = Field(..., alias="dealDate", description="Transaction date (ISO format)")
# Common optional fields
asset_area: Optional[float] = Field(None, alias="assetArea", description="Property area in sqm")
settlement_name_heb: Optional[str] = Field(None, alias="settlementNameHeb", description="City name in Hebrew")
property_type_description: Optional[str] = Field(None, alias="propertyTypeDescription", description="Property type")
neighborhood: Optional[str] = Field(None, description="Neighborhood name")
street_name: Optional[str] = Field(None, alias="streetName", description="Street name")
house_number: Optional[str] = Field(None, alias="houseNumber", description="House number")
# Floor information
floor: Optional[str] = Field(None, description="Floor description (may be Hebrew)")
floor_number: Optional[int] = Field(None, alias="floorNumber", description="Numeric floor number")
# Additional details
rooms: Optional[float] = Field(None, description="Number of rooms")
# Priority and metadata (added by our system, not from API)
priority: Optional[int] = Field(None, description="Priority for sorting (0=same building, 1=street, 2=neighborhood)")
# Geometry and internal fields (often not useful for analysis)
shape: Optional[str] = Field(None, description="WKT geometry")
source_polygon_id: Optional[str] = Field(None, alias="sourcePolygonId", description="Source polygon ID")
sourceorder: Optional[int] = Field(None, description="Source ordering")
model_config = ConfigDict(
populate_by_name=True, # Allow both alias and field name
extra='allow' # Allow extra fields from API that we don't model
)
@computed_field
@property
def price_per_sqm(self) -> Optional[float]:
"""
Calculated price per square meter.
Returns:
Price per sqm in NIS, or None if area is missing/zero
"""
if self.asset_area and self.asset_area > 0:
return round(self.deal_amount / self.asset_area, 2)
return None
@field_validator('deal_date', mode='before')
@classmethod
def parse_deal_date(cls, v: Any) -> str:
"""Parse deal date to ISO format string."""
if isinstance(v, str):
return v
if isinstance(v, datetime):
return v.isoformat()
return str(v)
class DealStatistics(BaseModel):
"""
Statistical analysis of real estate deals.
Attributes:
total_deals: Total number of deals analyzed
price_statistics: Statistics for deal prices
area_statistics: Statistics for property areas
price_per_sqm_statistics: Statistics for price per sqm
property_type_distribution: Count by property type
date_range: Earliest and latest deal dates
"""
total_deals: int = Field(..., description="Total number of deals analyzed")
# Price statistics
price_statistics: Dict[str, float] = Field(
default_factory=dict,
description="Price stats (mean, median, std_dev, min, max, percentiles)"
)
# Area statistics
area_statistics: Dict[str, float] = Field(
default_factory=dict,
description="Area stats (mean, median, std_dev, min, max)"
)
# Price per sqm statistics
price_per_sqm_statistics: Dict[str, float] = Field(
default_factory=dict,
description="Price/sqm stats (mean, median, std_dev, min, max)"
)
# Distribution by property type
property_type_distribution: Dict[str, int] = Field(
default_factory=dict,
description="Count of deals by property type"
)
# Date range
date_range: Optional[Dict[str, str]] = Field(
None,
description="Earliest and latest deal dates"
)
class MarketActivityScore(BaseModel):
"""
Market activity scoring metrics.
Attributes:
activity_score: Overall activity score (0-100)
total_deals: Total number of deals in period
deals_per_month: Average deals per month
trend: Market trend (increasing, stable, decreasing)
time_period_months: Analysis period in months
monthly_distribution: Deals per month breakdown
"""
activity_score: float = Field(..., description="Overall activity score (0-100)", ge=0, le=100)
total_deals: int = Field(..., description="Total deals in period")
deals_per_month: float = Field(..., description="Average deals per month")
trend: str = Field(..., description="Market trend (increasing, stable, decreasing)")
time_period_months: int = Field(..., description="Analysis period in months")
monthly_distribution: Dict[str, int] = Field(
default_factory=dict,
description="Deals per month (YYYY-MM: count)"
)
class InvestmentAnalysis(BaseModel):
"""
Investment potential analysis metrics.
Attributes:
investment_score: Overall investment score (0-100)
price_trend: Price trend direction
price_appreciation_rate: Annualized price appreciation rate (%)
price_volatility: Price volatility score (0-100, lower is more stable)
market_stability: Stability rating description
avg_price_per_sqm: Average price per square meter
price_change_pct: Total price change percentage
total_deals: Total deals analyzed (sample size)
data_quality: Data quality assessment
"""
investment_score: float = Field(..., description="Overall investment score (0-100)", ge=0, le=100)
price_trend: str = Field(..., description="Price trend (increasing, stable, decreasing)")
price_appreciation_rate: float = Field(..., description="Annual price growth rate (%)")
price_volatility: float = Field(..., description="Price volatility score (0-100)", ge=0, le=100)
market_stability: str = Field(..., description="Market stability rating")
avg_price_per_sqm: float = Field(..., description="Average price per sqm")
price_change_pct: float = Field(..., description="Total price change percentage")
total_deals: int = Field(..., description="Total deals analyzed (sample size)")
data_quality: str = Field(..., description="Data quality (excellent, good, fair, limited)")
class LiquidityMetrics(BaseModel):
"""
Market liquidity metrics.
Attributes:
liquidity_score: Overall liquidity score (0-100, based on velocity)
total_deals: Total deals in period
time_period_months: Analysis period in months
avg_deals_per_month: Average deals per month
liquidity_rating: Market liquidity rating
trend_direction: Liquidity trend direction
"""
liquidity_score: float = Field(..., description="Overall liquidity score (0-100)", ge=0, le=100)
total_deals: int = Field(..., description="Total deals in period")
time_period_months: int = Field(..., description="Analysis period in months")
avg_deals_per_month: float = Field(..., description="Average deals per month")
deal_velocity: float = Field(..., description="Deal velocity (deals per month)")
market_activity_level: str = Field(..., description="Activity level (very_high, high, moderate, low, very_low)")
class DealFilters(BaseModel):
"""
Filtering criteria for real estate deals.
All fields are optional - only specified filters are applied.
Attributes:
property_type: Filter by property type (דירה, בית, etc.)
min_rooms: Minimum number of rooms
max_rooms: Maximum number of rooms
min_price: Minimum deal amount (NIS)
max_price: Maximum deal amount (NIS)
min_area: Minimum asset area (sqm)
max_area: Maximum asset area (sqm)
min_floor: Minimum floor number
max_floor: Maximum floor number
"""
property_type: Optional[str] = Field(None, description="Property type filter")
min_rooms: Optional[float] = Field(None, description="Minimum rooms", ge=0)
max_rooms: Optional[float] = Field(None, description="Maximum rooms", ge=0)
min_price: Optional[float] = Field(None, description="Minimum price (NIS)", ge=0)
max_price: Optional[float] = Field(None, description="Maximum price (NIS)", ge=0)
min_area: Optional[float] = Field(None, description="Minimum area (sqm)", ge=0)
max_area: Optional[float] = Field(None, description="Maximum area (sqm)", ge=0)
min_floor: Optional[int] = Field(None, description="Minimum floor")
max_floor: Optional[int] = Field(None, description="Maximum floor")
@field_validator('max_rooms')
@classmethod
def validate_max_rooms(cls, v: Optional[float], info) -> Optional[float]:
"""Ensure max_rooms >= min_rooms if both specified."""
if v is not None and info.data.get('min_rooms') is not None:
if v < info.data['min_rooms']:
raise ValueError("max_rooms must be >= min_rooms")
return v
@field_validator('max_price')
@classmethod
def validate_max_price(cls, v: Optional[float], info) -> Optional[float]:
"""Ensure max_price >= min_price if both specified."""
if v is not None and info.data.get('min_price') is not None:
if v < info.data['min_price']:
raise ValueError("max_price must be >= min_price")
return v
@field_validator('max_area')
@classmethod
def validate_max_area(cls, v: Optional[float], info) -> Optional[float]:
"""Ensure max_area >= min_area if both specified."""
if v is not None and info.data.get('min_area') is not None:
if v < info.data['min_area']:
raise ValueError("max_area must be >= min_area")
return v
@field_validator('max_floor')
@classmethod
def validate_max_floor(cls, v: Optional[int], info) -> Optional[int]:
"""Ensure max_floor >= min_floor if both specified."""
if v is not None and info.data.get('min_floor') is not None:
if v < info.data['min_floor']:
raise ValueError("max_floor must be >= min_floor")
return v
nadlan_mcp/govmap/statistics.py
+76 -35
View File
@@ -5,18 +5,21 @@ This module provides pure mathematical functions for analyzing real estate deal
"""
from collections import Counter
from typing import Any, Dict, List
from typing import List
from datetime import datetime
from .models import Deal, DealStatistics
def calculate_deal_statistics(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
def calculate_deal_statistics(deals: List[Deal]) -> DealStatistics:
"""
Calculate statistical aggregations on deal data.
Args:
deals: List of deal dictionaries
deals: List of Deal model instances
Returns:
Dictionary with statistical metrics
DealStatistics model with comprehensive metrics
Raises:
ValueError: If deals is not a valid list
@@ -25,46 +28,52 @@ def calculate_deal_statistics(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
raise ValueError("deals must be a list")
if not deals:
return {
"count": 0,
"price_stats": {},
"area_stats": {},
"price_per_sqm_stats": {},
"room_distribution": {},
}
return DealStatistics(
total_deals=0,
price_statistics={},
area_statistics={},
price_per_sqm_statistics={},
property_type_distribution={},
date_range=None,
)
# Extract numeric values
prices = []
areas = []
price_per_sqm_values = []
rooms = []
property_types = []
deal_dates = []
for deal in deals:
price = deal.get("dealAmount")
if isinstance(price, (int, float)) and price > 0:
prices.append(price)
# Prices
if deal.deal_amount and deal.deal_amount > 0:
prices.append(deal.deal_amount)
area = deal.get("assetArea")
if isinstance(area, (int, float)) and area > 0:
areas.append(area)
# Areas
if deal.asset_area and deal.asset_area > 0:
areas.append(deal.asset_area)
pps = deal.get("price_per_sqm")
if pps is None and price and area and area > 0:
pps = price / area
if isinstance(pps, (int, float)) and pps > 0:
price_per_sqm_values.append(pps)
# Price per sqm (use computed field)
if deal.price_per_sqm:
price_per_sqm_values.append(deal.price_per_sqm)
room_count = deal.get("assetRoomNum")
if isinstance(room_count, (int, float)):
rooms.append(room_count)
# Property types
if deal.property_type_description:
property_types.append(deal.property_type_description)
# Deal dates
if deal.deal_date:
deal_dates.append(deal.deal_date)
# Calculate statistics
stats: Dict[str, Any] = {"count": len(deals)}
price_stats = {}
area_stats = {}
price_per_sqm_stats = {}
# Price statistics
if prices:
sorted_prices = sorted(prices)
stats["price_stats"] = {
price_stats = {
"mean": round(sum(prices) / len(prices), 2),
"median": (sorted_prices[len(sorted_prices) // 2] + sorted_prices[(len(sorted_prices) - 1) // 2]) / 2,
"min": min(prices),
@@ -78,7 +87,7 @@ def calculate_deal_statistics(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
# Area statistics
if areas:
sorted_areas = sorted(areas)
stats["area_stats"] = {
area_stats = {
"mean": round(sum(areas) / len(areas), 2),
"median": sorted_areas[len(sorted_areas) // 2],
"min": min(areas),
@@ -90,7 +99,7 @@ def calculate_deal_statistics(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
# Price per sqm statistics
if price_per_sqm_values:
sorted_pps = sorted(price_per_sqm_values)
stats["price_per_sqm_stats"] = {
price_per_sqm_stats = {
"mean": round(sum(price_per_sqm_values) / len(price_per_sqm_values), 2),
"median": round(sorted_pps[len(sorted_pps) // 2], 2),
"min": round(min(price_per_sqm_values), 2),
@@ -99,12 +108,44 @@ def calculate_deal_statistics(deals: List[Dict[str, Any]]) -> Dict[str, Any]:
"p75": round(sorted_pps[(3 * len(sorted_pps)) // 4], 2),
}
# Room distribution
if rooms:
room_counts = Counter(rooms)
stats["room_distribution"] = dict(sorted(room_counts.items()))
# Property type distribution
property_type_dist = {}
if property_types:
type_counts = Counter(property_types)
property_type_dist = dict(sorted(type_counts.items()))
return stats
# Date range
date_range_dict = None
if deal_dates:
try:
# Parse ISO date strings to get earliest and latest
parsed_dates = []
for date_str in deal_dates:
try:
# Handle ISO format with timezone (e.g., "2025-01-01T00:00:00.000Z")
if 'T' in date_str:
date_str = date_str.split('T')[0]
parsed_dates.append(date_str)
except:
continue
if parsed_dates:
sorted_dates = sorted(parsed_dates)
date_range_dict = {
"earliest": sorted_dates[0],
"latest": sorted_dates[-1],
}
except:
pass
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,
)
def calculate_std_dev(values: List[float]) -> float: