diff --git a/nadlan_mcp/fastmcp_server.py b/nadlan_mcp/fastmcp_server.py index e7f8df0..b809aea 100644 --- a/nadlan_mcp/fastmcp_server.py +++ b/nadlan_mcp/fastmcp_server.py @@ -85,25 +85,55 @@ def get_deals_by_radius(latitude: float, longitude: float, radius_meters: int = return f"Error fetching deals by radius: {str(e)}" @mcp.tool() -def get_street_deals(polygon_id: str, limit: int = 100) -> str: +def get_street_deals(polygon_id: str, limit: int = 100, deal_type: int = 2) -> str: """Get real estate deals for a specific street polygon. Args: polygon_id: The polygon ID of the street/area limit: Maximum number of deals to return (default: 100) + deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) Returns: JSON string containing recent real estate deals for the street """ try: - deals = client.get_street_deals(polygon_id, limit) + deals = client.get_street_deals(polygon_id, limit, deal_type=deal_type) if not deals: - return f"No deals found for polygon ID {polygon_id}" + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" + return f"No {deal_type_desc} deals found for polygon ID {polygon_id}" + # Add price per sqm calculation for each deal + for deal in 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 info + deal['deal_type'] = deal_type + deal['deal_type_description'] = 'first_hand_new' if deal_type == 1 else 'second_hand_used' + + # Calculate basic statistics including price per sqm + prices = [deal.get("dealAmount", 0) for deal in deals if deal.get("dealAmount")] + price_per_sqm_values = [deal.get("price_per_sqm", 0) for deal in deals if deal.get("price_per_sqm")] + + stats = {} + if price_per_sqm_values: + stats["price_per_sqm_stats"] = { + "average_price_per_sqm": round(sum(price_per_sqm_values) / len(price_per_sqm_values), 0), + "min_price_per_sqm": round(min(price_per_sqm_values), 0), + "max_price_per_sqm": round(max(price_per_sqm_values), 0) + } + + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" return json.dumps({ "total_deals": len(deals), "polygon_id": polygon_id, + "deal_type": deal_type, + "deal_type_description": deal_type_desc, + "market_statistics": stats, "deals": deals }, ensure_ascii=False, indent=2) @@ -112,7 +142,7 @@ def get_street_deals(polygon_id: str, limit: int = 100) -> str: return f"Error fetching street deals: {str(e)}" @mcp.tool() -def find_recent_deals_for_address(address: str, years_back: int = 2, radius_meters: int = 30, max_deals: int = 50) -> str: +def find_recent_deals_for_address(address: str, years_back: int = 2, radius_meters: int = 30, max_deals: int = 50, deal_type: int = 2) -> str: """Find recent real estate deals for a specific address. Args: @@ -121,15 +151,17 @@ def find_recent_deals_for_address(address: str, years_back: int = 2, radius_mete radius_meters: Search radius in meters from the address (default: 30) Small radius since street deals cover the entire street anyway max_deals: Maximum number of deals to return (default: 50, optimized for LLM token limits) + deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) Returns: JSON string containing recent real estate deals for the address """ try: - deals = client.find_recent_deals_for_address(address, years_back, radius_meters, max_deals) + deals = client.find_recent_deals_for_address(address, years_back, radius_meters, max_deals, deal_type) if not deals: - return f"No deals found for address '{address}'" + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" + return f"No {deal_type_desc} deals found for address '{address}'" # Calculate comprehensive statistics prices = [deal.get("dealAmount", 0) for deal in deals if deal.get("dealAmount")] @@ -178,12 +210,15 @@ def find_recent_deals_for_address(address: str, years_back: int = 2, radius_mete "median_price_per_sqm": round(sorted(price_per_sqm_values)[len(price_per_sqm_values)//2], 0) if price_per_sqm_values else 0 } + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" return json.dumps({ "search_parameters": { "address": address, "years_back": years_back, "radius_meters": radius_meters, - "max_deals": max_deals + "max_deals": max_deals, + "deal_type": deal_type, + "deal_type_description": deal_type_desc }, "market_statistics": stats, "deals": deals @@ -194,21 +229,23 @@ def find_recent_deals_for_address(address: str, years_back: int = 2, radius_mete return f"Error analyzing address: {str(e)}" @mcp.tool() -def get_neighborhood_deals(polygon_id: str, limit: int = 100) -> str: +def get_neighborhood_deals(polygon_id: str, limit: int = 100, deal_type: int = 2) -> str: """Get real estate deals for a specific neighborhood polygon. Args: polygon_id: The polygon ID of the neighborhood limit: Maximum number of deals to return (default: 100) + deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) Returns: JSON string containing recent real estate deals in the specified neighborhood """ try: - deals = client.get_neighborhood_deals(polygon_id, limit) + deals = client.get_neighborhood_deals(polygon_id, limit, deal_type=deal_type) if not deals: - return f"No deals found for polygon ID {polygon_id}" + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" + return f"No {deal_type_desc} deals found for polygon ID {polygon_id}" # Add price per sqm calculation for each deal for deal in deals: @@ -218,6 +255,9 @@ def get_neighborhood_deals(polygon_id: str, limit: int = 100) -> str: deal['price_per_sqm'] = round(price / area, 2) else: deal['price_per_sqm'] = None + # Add deal type info + deal['deal_type'] = deal_type + deal['deal_type_description'] = 'first_hand_new' if deal_type == 1 else 'second_hand_used' # Calculate basic statistics including price per sqm prices = [deal.get("dealAmount", 0) for deal in deals if deal.get("dealAmount")] @@ -231,9 +271,12 @@ def get_neighborhood_deals(polygon_id: str, limit: int = 100) -> str: "max_price_per_sqm": round(max(price_per_sqm_values), 0) } + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" return json.dumps({ "total_deals": len(deals), "polygon_id": polygon_id, + "deal_type": deal_type, + "deal_type_description": deal_type_desc, "market_statistics": stats, "deals": deals }, ensure_ascii=False, indent=2) @@ -243,24 +286,26 @@ def get_neighborhood_deals(polygon_id: str, limit: int = 100) -> str: return f"Error fetching neighborhood deals: {str(e)}" @mcp.tool() -def analyze_market_trends(address: str, years_back: int = 3, radius_meters: int = 300, max_deals: int = 100) -> str: +def analyze_market_trends(address: str, years_back: int = 3, radius_meters: int = 100, max_deals: int = 100, deal_type: int = 2) -> str: """Analyze market trends and price patterns for an area with comprehensive data. Args: address: The address to analyze trends around years_back: How many years of data to analyze (default: 3) - radius_meters: Search radius in meters from the address (default: 300, larger for trend analysis) + radius_meters: Search radius in meters from the address (default: 100, larger for trend analysis) max_deals: Maximum number of deals to analyze (default: 100, optimized for performance and token limits) + deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2) Returns: JSON string containing comprehensive market trend analysis (summarized data, not raw deals) """ try: # Get deals for the address with larger radius for trend analysis - deals = client.find_recent_deals_for_address(address, years_back, radius_meters, max_deals) + deals = client.find_recent_deals_for_address(address, years_back, radius_meters, max_deals, deal_type) if not deals: - return f"No deals found for comprehensive market analysis near '{address}'" + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" + return f"No {deal_type_desc} deals found for comprehensive market analysis near '{address}'" # Efficient analysis with reduced complexity from collections import defaultdict @@ -354,13 +399,17 @@ def analyze_market_trends(address: str, years_back: int = 3, radius_meters: int "last_year_avg_price_per_sqm": last_year['avg_price_per_sqm'] } + deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)" + # Return summarized analysis (NO raw deals to save tokens) return json.dumps({ "analysis_parameters": { "address": address, "years_analyzed": years_back, "radius_meters": radius_meters, - "deals_analyzed": len(deals) + "deals_analyzed": len(deals), + "deal_type": deal_type, + "deal_type_description": deal_type_desc }, "market_summary": { "total_deals": len(deals), diff --git a/nadlan_mcp/govmap.py b/nadlan_mcp/govmap.py index e58a8ef..7ca2d04 100644 --- a/nadlan_mcp/govmap.py +++ b/nadlan_mcp/govmap.py @@ -140,7 +140,8 @@ class GovmapClient: return [] def get_street_deals(self, polygon_id: str, limit: int = 10, - start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict[str, Any]]: + 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. @@ -149,6 +150,7 @@ class GovmapClient: 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 @@ -158,14 +160,14 @@ class GovmapClient: """ url = f"{self.base_url}/real-estate/street-deals/{polygon_id}" - params: Dict[str, Any] = {"limit": limit} + params: Dict[str, Any] = {"limit": limit, "dealType": deal_type} if start_date: params["startDate"] = start_date if end_date: params["endDate"] = end_date try: - logger.info(f"Getting street deals for polygon: {polygon_id}") + logger.info(f"Getting street deals for polygon: {polygon_id}, dealType: {deal_type}") response = self.session.get(url, params=params) response.raise_for_status() @@ -183,7 +185,8 @@ class GovmapClient: return [] def get_neighborhood_deals(self, polygon_id: str, limit: int = 10, - start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict[str, Any]]: + 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. @@ -192,6 +195,7 @@ class GovmapClient: 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 @@ -201,14 +205,14 @@ class GovmapClient: """ url = f"{self.base_url}/real-estate/neighborhood-deals/{polygon_id}" - params: Dict[str, Any] = {"limit": limit} + params: Dict[str, Any] = {"limit": limit, "dealType": deal_type} if start_date: params["startDate"] = start_date if end_date: params["endDate"] = end_date try: - logger.info(f"Getting neighborhood deals for polygon: {polygon_id}") + logger.info(f"Getting neighborhood deals for polygon: {polygon_id}, dealType: {deal_type}") response = self.session.get(url, params=params) response.raise_for_status() @@ -226,7 +230,8 @@ class GovmapClient: return [] def find_recent_deals_for_address(self, address: str, years_back: int = 2, - radius: int = 30, max_deals: int = 50) -> List[Dict[str, Any]]: + radius: int = 30, max_deals: int = 50, + deal_type: int = 2) -> List[Dict[str, Any]]: """ Find all relevant real estate deals for a given address from the last few years. @@ -238,7 +243,8 @@ class GovmapClient: 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: 200) + max_deals: Maximum number of deals to return (default: 50) + 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, @@ -250,7 +256,7 @@ class GovmapClient: """ try: # Step 1: Get coordinates for the address - logger.info(f"Starting search for address: {address}") + logger.info(f"Starting search for address: {address}, dealType: {deal_type}") autocomplete_result = self.autocomplete_address(address) if not autocomplete_result.get('results'): @@ -306,13 +312,13 @@ class GovmapClient: # 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 + 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 + start_date=start_date_str, end_date=end_date_str, deal_type=deal_type ) # Process street deals and separate building deals @@ -359,7 +365,7 @@ class GovmapClient: if len(all_deals) > max_deals: all_deals = all_deals[:max_deals] - # Add price per square meter calculation + # 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) @@ -367,9 +373,14 @@ class GovmapClient: 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"(Building: {len(building_deals)}, Street: {len(street_deals)}, Neighborhood: {len(neighborhood_deals)}) " + f"[{deal['deal_type_description'] if all_deals else 'N/A'}]") return all_deals except Exception as e: