diff --git a/.github/workflows/ci-quality.yml b/.github/workflows/ci-quality.yml index c0d04aaa..a3b24f41 100644 --- a/.github/workflows/ci-quality.yml +++ b/.github/workflows/ci-quality.yml @@ -222,6 +222,59 @@ jobs: - name: Mypy run: mypy core/ investor_app/finance/ --ignore-missing-imports --disable-error-code var-annotated --disable-error-code attr-defined --disable-error-code operator --disable-error-code misc --disable-error-code has-type --disable-error-code arg-type --disable-error-code assignment + # ── Lighthouse PWA audit ──────────────────────────────────────────────── + + lighthouse: + name: "🚀 Lighthouse PWA Audit" + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: ./.github/actions/python-setup + - name: Install system deps + run: sudo apt-get install -y libcairo2-dev + - name: Install dependencies + run: pip install -r requirements.txt + - name: Collect static files + run: python manage.py collectstatic --noinput --verbosity 0 + - name: Install Lighthouse CI + run: npm install -g @lhci/cli + - name: Run Lighthouse CI + continue-on-error: true + run: | + export SECRET_KEY=$(python -c 'import secrets; print(secrets.token_hex(50))') + export DATABASE_URL="sqlite:///tmp/lighthouse_test.db" + export RUN_MIGRATIONS=1 + export SKIP_SEED=1 + export DJANGO_ENV=development + export DEBUG=True + export ALLOWED_HOSTS=localhost,127.0.0.1 + + # Start Django server in background + python manage.py runserver 0.0.0.0:8000 & + SERVER_PID=$! + + # Wait for server to start + for i in $(seq 1 30); do + if curl -s http://localhost:8000/health/ > /dev/null 2>&1; then + echo "Server started" + break + fi + sleep 2 + done + + # Run Lighthouse CI + lhci autorun --config=lighthouserc.json --collect.staticDistDir=./staticfiles + + # Stop server + kill $SERVER_PID 2>/dev/null || true + - name: Upload Lighthouse results + uses: actions/upload-artifact@v7 + if: always() + with: + name: lighthouse-results + path: .lighthouseci/ + retention-days: 7 + # ── Acceptance tests (real HTTP, via pytest-django live_server) ──────────── # BASE_URL is intentionally unset here: tests/acceptance/conftest.py falls # back to a live_server for this run. Full acceptance tests also run diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml index ee44f162..0eef715c 100644 --- a/.github/workflows/docker-publish.yml +++ b/.github/workflows/docker-publish.yml @@ -261,7 +261,7 @@ jobs: run: | echo "job-start: $(date -u +%Y-%m-%dT%H:%M:%SZ)" echo "sha: ${{ github.sha }}" - - name: Trivy — block HIGH/CRITICAL CVEs with fixes + - name: Trivy — block HIGH/CRITICAL CVEs with retries uses: aquasecurity/trivy-action@v0.36.0 with: scan-type: image @@ -271,6 +271,9 @@ jobs: severity: CRITICAL,HIGH ignore-unfixed: true vuln-type: os,library + version: v0.74.0 + timeout: 10m + skip-version-check: true - name: Trivy — upload SARIF uses: aquasecurity/trivy-action@v0.36.0 if: always() @@ -281,6 +284,9 @@ jobs: output: trivy-tier-2.sarif severity: CRITICAL,HIGH,MEDIUM ignore-unfixed: true + version: v0.74.0 + timeout: 10m + skip-version-check: true - name: Upload SARIF results uses: github/codeql-action/upload-sarif@v4 if: always() diff --git a/core/integrations/market/market_trends.py b/core/integrations/market/market_trends.py new file mode 100644 index 00000000..0712a74e --- /dev/null +++ b/core/integrations/market/market_trends.py @@ -0,0 +1,369 @@ +"""Market trends adapter for fetching and classifying market cycle indicators. + +This module provides adapters for various market data sources and +classification logic for determining market health status. +""" + +from __future__ import annotations + +import logging +from decimal import Decimal +from typing import Any, Dict, List, Optional +from dataclasses import dataclass +from datetime import date + +logger = logging.getLogger(__name__) + + +@dataclass +class MarketIndicatorData: + """Data class for market indicator values.""" + + indicator_type: str + value: Decimal + date_recorded: date + source: str = "" + notes: str = "" + + +def classify_market_health( + indicator_type: str, + value: Decimal, + metro_area: str = "", + median_income: Optional[Decimal] = None, +) -> str: + """Classify market health status for a given indicator. + + Args: + indicator_type: Type of indicator (median_price, dom, months_supply, etc.) + value: The indicator value + metro_area: Metropolitan area name (optional, for future context) + median_income: Median household income for price-to-income calculations + + Returns: + str: 'healthy', 'caution', or 'overheated' + """ + if indicator_type == "median_price": + if median_income is None or median_income <= 0: + return "caution" # Can't determine without income + ratio = value / median_income + if ratio <= Decimal("4.0"): + return "healthy" + elif ratio <= Decimal("5.0"): + return "caution" + else: + return "overheated" + + elif indicator_type == "dom": + # Days on Market: lower = hotter market + # < 10 days = overheated, 10-30 days = caution, > 30 days = healthy + if value < Decimal("10"): + return "overheated" + elif value < Decimal("30"): + return "caution" + else: + return "healthy" + + elif indicator_type == "months_supply": + # Months of supply: lower = seller's market (overheated) + if value < Decimal("3"): + return "overheated" + elif value <= Decimal("6"): + return "healthy" + else: + return "caution" # > 6 months = buyer's market (declining) + + elif indicator_type == "price_to_income": + # Price-to-income ratio + if value <= Decimal("4.0"): + return "healthy" + elif value <= Decimal("5.0"): + return "caution" + else: + return "overheated" + + elif indicator_type == "rent_growth_yoy": + # Year-over-year rent growth + if value < Decimal("0"): + return "caution" # Declining + elif value <= Decimal("0.05"): # 0-5% + return "healthy" + elif value <= Decimal("0.08"): # 5-8% + return "caution" + else: + return "overheated" # > 8% + + else: + return "caution" + + +def fetch_market_indicators(metro_area: str) -> List[Dict[str, Any]]: + """Fetch market indicators for a metro area. + + This is a placeholder implementation. In production, this would + fetch from Zillow Research API, Census, BLS, FRED, etc. + + Args: + metro_area: Metropolitan Statistical Area name + + Returns: + List of indicator data dictionaries + """ + # Placeholder implementation - returns mock data for testing + # In production, this would call external APIs + from datetime import date + from decimal import Decimal + + today = date.today() + return [ + { + "indicator_type": "median_price", + "value": Decimal("425000"), + "date_recorded": today, + "source": "zillow", + "notes": "Zillow Home Value Index", + }, + { + "indicator_type": "dom", + "value": Decimal("28"), + "date_recorded": today, + "source": "zillow", + "notes": "Zillow Days on Market", + }, + { + "indicator_type": "months_supply", + "value": Decimal("3.5"), + "date_recorded": today, + "source": "zillow", + "notes": "Zillow Months of Supply", + }, + { + "indicator_type": "price_to_income", + "value": Decimal("4.2"), + "date_recorded": today, + "source": "census", + "notes": "Census ACS median price / median income", + }, + { + "indicator_type": "rent_growth_yoy", + "value": Decimal("0.045"), + "date_recorded": today, + "source": "zillow", + "notes": "Zillow Observed Rent Index YoY", + }, + ] + + +def get_indicator_history( + metro_area: str, indicator_type: str, limit: int = 12 +) -> List[Decimal]: + """Return up to ``limit`` most recent values for an indicator, oldest first.""" + from core.models.growth import MarketIndicator + + qs = MarketIndicator.objects.filter( + metro_area=metro_area, + indicator_type=indicator_type, + ).order_by("-date_recorded")[:limit] + return [ind.value for ind in reversed(list(qs))] + + +def build_sparkline_points( + values: List[Decimal], width: int = 100, height: int = 30 +) -> str: + """Convert a value series into SVG polyline points for a sparkline. + + Points are normalized to the ``width`` x ``height`` viewBox. A flat line + is drawn when all values are equal, and the midpoint when empty/constant. + + Args: + values: Series of Decimal values, oldest first. + width: ViewBox width (default 100). + height: ViewBox height (default 30). + + Returns: + Space-separated "x,y" point string suitable for . + """ + mid = height // 2 + if not values: + return "" + if len(values) == 1: + return f"0,{mid} {width},{mid}" + + v = [float(x) for x in values] + lo, hi = min(v), max(v) + rng = hi - lo + if rng == 0: + step = width / (len(v) - 1) + return " ".join(f"{int(round(i * step))},{mid}" for i in range(len(v))) + + step = width / (len(v) - 1) + pts = [] + for i, val in enumerate(v): + x = int(round(i * step)) + y = int(round(height - (val - lo) / rng * height)) + pts.append(f"{x},{y}") + return " ".join(pts) + + +def get_latest_indicators(metro_area: str) -> Dict[str, Any]: + """Get latest indicators for a metro area. + + Stored ``MarketIndicator`` records take precedence; the adapter + (``fetch_market_indicators``) fills in any indicator type with no + stored record yet. + + Args: + metro_area: Metropolitan Statistical Area name + + Returns: + Dictionary mapping indicator types to their latest values, + health classification, history, and sparkline points. + """ + from core.models.growth import MarketIndicator, MarketIndicatorType + + # Median income context improves median_price classification. + median_income = None + try: + from core.models.growth import MarketSnapshot + + snapshot = MarketSnapshot.objects.filter( + msa_name__icontains=metro_area.split(",")[0].strip() + ).first() + if snapshot and snapshot.median_household_income: + median_income = snapshot.median_household_income + except Exception: + logger.warning( + "get_latest_indicators: MarketSnapshot median-income lookup failed " + "for %s; classifying without income context", + metro_area, + exc_info=True, + ) + median_income = None + + result: Dict[str, Any] = {} + fetched: Dict[str, Dict[str, Any]] = { + ind["indicator_type"]: ind for ind in fetch_market_indicators(metro_area) + } + + for itype in MarketIndicatorType.values: + latest = ( + MarketIndicator.objects.filter(metro_area=metro_area, indicator_type=itype) + .order_by("-date_recorded") + .first() + ) + if latest is not None: + value = latest.value + date_recorded = latest.date_recorded + source = latest.source + history = get_indicator_history(metro_area, itype) + elif itype in fetched: + value = fetched[itype]["value"] + date_recorded = fetched[itype]["date_recorded"] + source = fetched[itype].get("source", "") + history = [value] + else: + continue + + result[itype] = { + "value": value, + "date_recorded": date_recorded, + "source": source, + "health": classify_market_health( + itype, + value, + metro_area=metro_area, + median_income=median_income, + ), + "history": history, + "sparkline": build_sparkline_points(history), + } + return result + + +def update_market_indicators(metro_area: str = "") -> Dict[str, Any]: + """Update market indicators for one or all metro areas. + + This function would be called by the management command to + fetch and store the latest indicator values. + + Args: + metro_area: Specific metro area to update, or empty for all + + Returns: + Dict with stats about the update operation + """ + from core.models.growth import MarketIndicator + + if metro_area: + metro_areas = [metro_area] + else: + # Get unique metro areas from existing indicators + metro_areas = list( + MarketIndicator.objects.values_list("metro_area", flat=True).distinct() + ) + + created = 0 + updated = 0 + errors = 0 + + for metro in metro_areas: + try: + indicators = fetch_market_indicators(metro) + for ind_data in indicators: + obj, was_created = MarketIndicator.objects.update_or_create( + metro_area=metro, + indicator_type=ind_data["indicator_type"], + date_recorded=ind_data["date_recorded"], + defaults={ + "value": ind_data["value"], + "source": ind_data.get("source", ""), + "notes": ind_data.get("notes", ""), + }, + ) + if was_created: + created += 1 + else: + updated += 1 + except Exception: + errors += 1 + # Log error in production + pass + + return { + "created": created, + "updated": updated, + "errors": errors, + "metro_areas": metro_areas, + } + + +def get_market_health_summary(metro_area: str) -> Dict[str, Any]: + """Get a summary of market health for a metro area. + + Args: + metro_area: Metropolitan Statistical Area name + + Returns: + Dictionary with overall health and per-indicator details + """ + indicators = get_latest_indicators(metro_area) + + health_counts = {"healthy": 0, "caution": 0, "overheated": 0} + for ind_type, data in indicators.items(): + health = data.get("health", "caution") + health_counts[health] += 1 + + # Determine overall health + if health_counts["overheated"] > health_counts["healthy"]: + overall = "overheated" + elif health_counts["caution"] > health_counts["healthy"]: + overall = "caution" + else: + overall = "healthy" + + return { + "metro_area": metro_area, + "overall_health": overall, + "health_counts": health_counts, + "indicators": indicators, + } diff --git a/core/integrations/sources/__init__.py b/core/integrations/sources/__init__.py index 7866c922..2e6fb6be 100644 --- a/core/integrations/sources/__init__.py +++ b/core/integrations/sources/__init__.py @@ -6,4 +6,11 @@ 'zip_code': '...', 'price': Decimal, 'beds': int, 'baths': Decimal, 'sq_ft': int, 'property_type': 'SFH', 'url': 'http...', 'posted_at': datetime } + +The RESO Web API adapter (MLS data feed) lives in +``core.integrations.sources.reso_adapter`` — import it directly: + + from core.integrations.sources.reso_adapter import RESOAdapter """ + +__all__ = ["reso_adapter", "attom_adapter", "dummy_adapter"] diff --git a/core/integrations/sources/reso_adapter.py b/core/integrations/sources/reso_adapter.py new file mode 100644 index 00000000..0507dfb2 --- /dev/null +++ b/core/integrations/sources/reso_adapter.py @@ -0,0 +1,791 @@ +"""RESO Web API adapter for MLS data feed integration. + +Implements the RESO Web API standard for MLS data feed integration, +supporting OData v4 queries, authentication, pagination, and data normalization. + +RESO Web API Specification: https://www.reso.org/web-api/ +""" + +from __future__ import annotations + +import hashlib +import logging +import os +import time +from datetime import datetime, timedelta +from decimal import Decimal +from typing import Any, Dict, List, Optional, Iterator +from urllib.parse import urlencode, urljoin + +import requests +from django.core.cache import cache +from django.utils import timezone + +logger = logging.getLogger(__name__) + + +class RESOAPIError(Exception): + """Base exception for RESO Web API errors.""" + + pass + + +class RESOAuthenticationError(RESOAPIError): + """Authentication failed with RESO Web API.""" + + pass + + +class RESORateLimitError(RESOAPIError): + """Rate limit exceeded for RESO Web API.""" + + pass + + +class RESOAdapter: + """ + Adapter for RESO Web API (MLS data feed). + + Implements OData v4 queries for MLS data feed integration, + supporting Property, Member, Office, and Media resources. + + RESO Web API Specification: https://www.reso.org/web-api/ + """ + + # Standard RESO Web API endpoints + DEFAULT_BASE_URL = "https://api.mls.example.com/odata" + + # Standard RESO resource types + RESOURCE_TYPES = [ + "Property", + "Member", + "Office", + "Media", + "OpenHouse", + "Room", + "Unit", + ] + + # Standard OData query options + ODATA_QUERY_OPTIONS = [ + "$filter", + "$select", + "$expand", + "$orderby", + "$top", + "$skip", + "$count", + "$skip", + "$top", + "$inlinecount", + ] + + # Cache duration for property details (12 hours) + CACHE_DURATION = 43200 # 12 hours in seconds + MAX_RETRIES = 3 + REQUEST_TIMEOUT = 30 # seconds + + def __init__( + self, + base_url: Optional[str] = None, + username: Optional[str] = None, + password: Optional[str] = None, + access_token: Optional[str] = None, + client_id: Optional[str] = None, + client_secret: Optional[str] = None, + token_url: Optional[str] = None, + ): + """ + Initialize RESO Web API adapter. + + Args: + base_url: RESO Web API base URL (OData service endpoint) + username: Username for Basic Auth (optional) + password: Password for Basic Auth (optional) + access_token: Bearer token for OAuth2 (optional) + client_id: OAuth2 client ID (optional) + client_secret: OAuth2 client secret (optional) + token_url: OAuth2 token endpoint URL (optional) + """ + self.base_url = base_url or os.getenv( + "RESO_API_BASE_URL", self.DEFAULT_BASE_URL + ) + self.username = username or os.getenv("RESO_USERNAME") + self.password = password or os.getenv("RESO_PASSWORD") + self.access_token = access_token or os.getenv("RESO_ACCESS_TOKEN") + self.client_id = client_id or os.getenv("RESO_CLIENT_ID") + self.client_secret = client_secret or os.getenv("RESO_CLIENT_SECRET") + self.token_url = token_url or os.getenv("RESO_TOKEN_URL") + + self.session = requests.Session() + self._access_token: Optional[str] = self.access_token + self._token_expiry: Optional[datetime] = None + + self._setup_auth() + + def _setup_auth(self) -> None: + """Configure authentication headers for the session.""" + if self.access_token: + self.session.headers.update( + {"Authorization": f"Bearer {self.access_token}"} + ) + elif self.username and self.password: + import base64 + + credentials = base64.b64encode( + f"{self.username}:{self.password}".encode() + ).decode() + self.session.headers.update({"Authorization": f"Basic {credentials}"}) + + self.session.headers.update( + { + "Accept": "application/json", + "Content-Type": "application/json", + } + ) + + def _get_access_token(self) -> Optional[str]: + """Obtain OAuth2 access token if client credentials are configured.""" + if not self.client_id or not self.client_secret or not self.token_url: + return None + + if self._access_token and self._token_expiry: + if timezone.now() < self._token_expiry - timedelta(minutes=5): + return self._access_token + + try: + response = requests.post( + self.token_url, + data={ + "grant_type": "client_credentials", + "client_id": self.client_id, + "client_secret": self.client_secret, + }, + timeout=10, + ) + response.raise_for_status() + token_data = response.json() + self._access_token = token_data.get("access_token") + expires_in = token_data.get("expires_in", 3600) + self._token_expiry = timezone.now() + timedelta(seconds=expires_in - 60) + return self._access_token + except Exception as e: + logger.error(f"Failed to obtain access token: {e}") + raise RESOAuthenticationError(f"Failed to obtain access token: {e}") + + def _get_headers(self) -> Dict[str, str]: + """Build request headers with current authentication.""" + headers: Dict[str, str] = { + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "PREI-RESO-Adapter/1.0", + } + + if self.access_token or self.client_id: + token = self._get_access_token() + if token: + headers["Authorization"] = f"Bearer {token}" + return headers + + return headers + + def _execute_request( + self, + method: str, + endpoint: str, + params: Optional[Dict[str, Any]] = None, + data: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """ + Execute HTTP request with retry logic and error handling. + + Args: + method: HTTP method (GET, POST, etc.) + endpoint: API endpoint path + params: Query parameters + data: Request body data + + Returns: + Parsed JSON response + + Raises: + RESOAuthenticationError: Authentication failed + RESORateLimitError: Rate limit exceeded + RESOAPIError: Other API errors + """ + url = urljoin(self.base_url or "", endpoint) + headers = self._get_headers() + + for attempt in range(self.MAX_RETRIES): + try: + response = self.session.request( + method=method, + url=url, + headers=headers, + params=params, + json=data, + timeout=self.REQUEST_TIMEOUT, + ) + + if response.status_code == 401: + # Token may be expired, force refresh + if self.access_token or self.client_id: + self._access_token = None + self._token_expiry = None + if self._get_access_token(): + continue + raise RESOAuthenticationError("Authentication failed") + + if response.status_code == 429: + retry_after = int(response.headers.get("Retry-After", "60")) + if attempt < self.MAX_RETRIES - 1: + logger.warning( + f"Rate limited, waiting {retry_after}s before retry" + ) + time.sleep(retry_after) + continue + raise RESORateLimitError("Rate limit exceeded") + + if not response.ok: + raise RESOAPIError( + f"API error: {response.status_code} - {response.text}" + ) + + result: Dict[str, Any] = response.json() + return result + + except requests.exceptions.Timeout: + if attempt == self.MAX_RETRIES - 1: + raise RESOAPIError("Request timeout") + time.sleep(2**attempt) # Exponential backoff + + except requests.exceptions.RequestException as e: + if attempt == self.MAX_RETRIES - 1: + raise RESOAPIError(f"Request failed: {e}") + time.sleep(2**attempt) + + raise RESOAPIError("Max retries exceeded") + + def _build_odata_query( + self, + filter_expr: Optional[str] = None, + select: Optional[List[str]] = None, + expand: Optional[List[str]] = None, + order_by: Optional[str] = None, + top: Optional[int] = None, + skip: int = 0, + count: bool = False, + ) -> Dict[str, str]: + """Build OData query parameters.""" + params: Dict[str, str] = {} + + if filter_expr: + params["$filter"] = filter_expr + if select: + params["$select"] = ",".join(select) + if expand: + params["$expand"] = ",".join(expand) + if order_by: + params["$orderby"] = order_by + if top is not None: + params["$top"] = str(top) + if skip: + params["$skip"] = str(skip) + if count: + params["$count"] = "true" + + return params + + def _build_filter( + self, + field: str, + operator: str, + value: Any, + ) -> str: + """Build OData filter expression.""" + rendered: Any = value + if isinstance(rendered, str): + rendered = f"'{rendered}'" + elif isinstance(rendered, datetime): + rendered = rendered.strftime("%Y-%m-%dT%H:%M:%SZ") + elif isinstance(rendered, bool): + rendered = str(rendered).lower() + return f"{field} {operator} {rendered}" + + def build_filter( + self, + filters: List[Dict[str, Any]], + ) -> str: + """Build complex OData filter from filter conditions. + + Args: + filters: List of filter dicts with keys: field, operator, value + Operators: eq, ne, gt, ge, lt, le, contains, startswith, endswith + + Returns: + OData filter string + """ + filter_parts = [] + for f in filters: + field = f["field"] + operator = f["operator"] + value = f["value"] + filter_parts.append(self._build_filter(field, operator, value)) + return " and ".join(filter_parts) + + def _get_cache_key(self, resource: str, params: Dict[str, Any]) -> str: + """Generate cache key for request.""" + key_data = f"{self.base_url}/{resource}?{urlencode(sorted(params.items()))}" + return hashlib.sha256(key_data.encode()).hexdigest() + + def _get_cached( + self, resource: str, params: Dict[str, Any] + ) -> Optional[Dict[str, Any]]: + """Get cached response if available.""" + cache_key = self._get_cache_key(resource, params) + cached: Optional[Dict[str, Any]] = cache.get(cache_key) + return cached + + def _set_cache(self, resource: str, params: Dict[str, Any], data: Dict) -> None: + """Cache response data.""" + cache_key = self._get_cache_key(resource, params) + cache.set(cache_key, data, timeout=self.CACHE_DURATION) + + # ─── Property Resource Methods ──────────────────────────────────────── + + def fetch_property( + self, + listing_id: str, + expand: Optional[List[str]] = None, + select: Optional[List[str]] = None, + ) -> Dict[str, Any]: + """ + Fetch a single property by ListingId. + + Args: + listing_id: MLS Listing ID (ListingKey) + expand: Navigation properties to expand (Media, OpenHouse, etc.) + select: Properties to return (OData $select) + + Returns: + Property data dictionary + """ + params = self._build_odata_query( + expand=expand, + select=select, + ) + endpoint = f"Property('{listing_id}')" + return self._execute_request("GET", endpoint, params=params) + + def query_properties( + self, + filter_expr: Optional[str] = None, + select: Optional[List[str]] = None, + expand: Optional[List[str]] = None, + order_by: Optional[str] = None, + top: int = 100, + skip: int = 0, + count: bool = False, + ) -> Dict[str, Any]: + """ + Query properties with OData filters. + + Args: + filter_expr: OData $filter expression + select: Properties to return ($select) + expand: Navigation properties to expand ($expand) + order_by: Sort order ($orderby) + top: Maximum results ($top) + skip: Records to skip ($skip) + count: Include total count ($count) + + Returns: + Dictionary with 'value' (list of properties) and optional '@odata.count' + """ + params = self._build_odata_query( + filter_expr=filter_expr, + select=select, + expand=expand, + order_by=order_by, + top=top, + skip=skip, + count=count, + ) + return self._execute_request("GET", "Property", params=params) + + def search_properties( + self, + city: Optional[str] = None, + state: Optional[str] = None, + postal_code: Optional[str] = None, + min_price: Optional[Decimal] = None, + max_price: Optional[Decimal] = None, + min_beds: Optional[int] = None, + max_beds: Optional[int] = None, + min_baths: Optional[Decimal] = None, + max_baths: Optional[Decimal] = None, + property_type: Optional[str] = None, + status: Optional[str] = None, + min_sqft: Optional[int] = None, + max_sqft: Optional[int] = None, + listing_status: Optional[str] = "Active", + days_on_market_max: Optional[int] = None, + top: int = 100, + skip: int = 0, + ) -> Dict[str, Any]: + """ + Search properties with common filters. + + Args: + city: City name + state: State abbreviation (e.g., "TX") + postal_code: ZIP code + min_price: Minimum list price + max_price: Maximum list price + min_beds: Minimum bedrooms + max_beds: Maximum bedrooms + min_baths: Minimum bathrooms + max_baths: Maximum bathrooms + property_type: Property type (SFH, Condo, etc.) + status: Listing status (Active, Pending, Closed) + min_sqft: Minimum square footage + max_sqft: Maximum square footage + listing_status: Listing status filter + days_on_market_max: Maximum days on market + top: Maximum results + skip: Records to skip + + Returns: + Dictionary with property listings + """ + filters: List[Dict[str, Any]] = [] + + if city: + filters.append({"field": "City", "operator": "eq", "value": city}) + if state: + filters.append( + {"field": "StateOrProvince", "operator": "eq", "value": state} + ) + if postal_code: + filters.append( + {"field": "PostalCode", "operator": "eq", "value": postal_code} + ) + if min_price is not None: + filters.append({"field": "ListPrice", "operator": "ge", "value": min_price}) + if max_price is not None: + filters.append({"field": "ListPrice", "operator": "le", "value": max_price}) + if min_beds is not None: + filters.append( + {"field": "BedroomsTotal", "operator": "ge", "value": min_beds} + ) + if max_beds is not None: + filters.append( + {"field": "BedroomsTotal", "operator": "le", "value": max_beds} + ) + if min_baths is not None: + filters.append( + {"field": "BathroomsTotalInteger", "operator": "ge", "value": min_baths} + ) + if max_baths is not None: + filters.append( + {"field": "BathroomsTotalInteger", "operator": "le", "value": max_baths} + ) + if property_type: + filters.append( + {"field": "PropertyType", "operator": "eq", "value": property_type} + ) + if listing_status: + filters.append( + {"field": "StandardStatus", "operator": "eq", "value": listing_status} + ) + if min_sqft is not None: + filters.append({"field": "LivingArea", "operator": "ge", "value": min_sqft}) + if max_sqft is not None: + filters.append({"field": "LivingArea", "operator": "le", "value": max_sqft}) + if days_on_market_max is not None: + filters.append( + {"field": "DaysOnMarket", "operator": "le", "value": days_on_market_max} + ) + + filter_expr = self.build_filter(filters) if filters else None + + return self.query_properties( + filter_expr=filter_expr, + top=top, + skip=skip, + ) + + def get_property_media( + self, + listing_id: str, + top: int = 50, + ) -> Dict[str, Any]: + """Fetch media (photos, videos) for a property.""" + return self.query_properties( + filter_expr=f"Media/any(m: m/ListingKey eq '{listing_id}')", + top=top, + ) + + # ─── Member/Office Methods ───────────────────────────────────────── + + def fetch_member( + self, + member_id: str, + expand: Optional[List[str]] = None, + ) -> Dict[str, Any]: + """Fetch agent/broker details by MemberKey.""" + params = self._build_odata_query(expand=expand) + endpoint = f"Member('{member_id}')" + return self._execute_request("GET", endpoint, params=params) + + def query_members( + self, + filter_expr: Optional[str] = None, + select: Optional[List[str]] = None, + top: int = 100, + skip: int = 0, + ) -> Dict[str, Any]: + """Query members (agents/brokers).""" + params = self._build_odata_query( + filter_expr=filter_expr, + select=select, + top=top, + skip=skip, + ) + return self._execute_request("GET", "Member", params=params) + + def fetch_office( + self, + office_id: str, + expand: Optional[List[str]] = None, + ) -> Dict[str, Any]: + """Fetch office details by OfficeKey.""" + params = self._build_odata_query(expand=expand) + endpoint = f"Office('{office_id}')" + return self._execute_request("GET", endpoint, params=params) + + # ─── Media/Media ─────────────────────────────────────────────────── + + def fetch_media( + self, + listing_id: str, + top: int = 50, + ) -> Dict[str, Any]: + """Fetch media (photos, videos) for a listing.""" + return self.query_properties( + filter_expr=f"Media/any(m: m/ListingKey eq '{listing_id}')", + select=[ + "MediaKey", + "MediaURL", + "MediaType", + "Order", + "MediaCategory", + "Description", + ], + top=top, + ) + + # ─── Pagination Helper ────────────────────────────────────────────── + + def iter_all( + self, + resource: str, + filter_expr: Optional[str] = None, + select: Optional[List[str]] = None, + batch_size: int = 100, + ) -> Iterator[Dict[str, Any]]: + """ + Iterate through all pages of a resource. + + Args: + resource: Resource name (Property, Member, etc.) + filter_expr: OData filter expression + select: Properties to select + batch_size: Page size + + Yields: + Individual resource records + """ + skip = 0 + while True: + result = self._execute_request( + "GET", + resource, + params=self._build_odata_query( + filter_expr=filter_expr, + top=batch_size, + skip=skip, + count=True, + ), + ) + + items = result.get("value", []) + if not items: + break + + for item in items: + yield item + + if len(items) < batch_size: + break + + skip += batch_size + + # ─── Cache Management ────────────────────────────────────────────── + + def clear_cache(self, pattern: Optional[str] = None) -> int: + """Clear cache entries matching pattern.""" + if pattern: + # Would need a more sophisticated cache backend for pattern deletion + logger.warning("Pattern-based cache clearing not fully implemented") + return 0 + cache.clear() + return 1 + + def get_cache_stats(self) -> Dict[str, Any]: + """Get cache statistics (requires django-redis or similar).""" + return { + "cache_backend": str(cache.__class__), + "cache_duration_seconds": self.CACHE_DURATION, + } + + # ─── Raw Query Support ────────────────────────────────────────────── + + def raw_query( + self, + resource: str, + odata_query: str, + ) -> Dict[str, Any]: + """ + Execute raw OData query string. + + Args: + resource: Resource name (Property, Member, etc.) + odata_query: Raw OData query string (e.g., "$filter=City eq 'Austin'&$top=10") + + Returns: + Query results + """ + params: Dict[str, Any] = {} + for pair in odata_query.split("&"): + if "=" in pair: + k, v = pair.split("=", 1) + if k in params: + existing = params[k] + if not isinstance(existing, list): + existing = [existing] + existing.append(v) + params[k] = existing + else: + params[k] = v + + return self._execute_request("GET", resource, params=params) + + +# ─── Utility Functions ───────────────────────────────────────────────── + + +def normalize_property_type(ptype: Optional[str]) -> Optional[str]: + """Normalize property type to standard values.""" + if not ptype: + return None + ptype = ptype.upper().strip() + mapping = { + "SINGLE FAMILY": "SFR", + "SINGLE FAMILY RESIDENCE": "SFR", + "SINGLE FAMILY DETACHED": "SFR", + "CONDO": "CONDO", + "CONDOMINIUM": "CONDO", + "TOWNHOUSE": "TOWNHOUSE", + "TOWN HOUSE": "TOWNHOUSE", + "DUPLEX": "DUPLEX", + "TRIPLEX": "TRIPLEX", + "FOURPLEX": "FOURPLEX", + "MULTIFAMILY": "MULTIFAMILY", + "APARTMENT": "APARTMENT", + "COMMERCIAL": "COMMERCIAL", + "LAND": "LAND", + "LOT": "LAND", + } + return mapping.get(ptype, ptype) + + +def normalize_property_data(raw: Dict[str, Any]) -> Dict[str, Any]: + """ + Normalize raw RESO property data to internal format. + + Args: + raw: Raw property data from RESO API + + Returns: + Normalized property data dict + """ + return { + "source": "reso", + "listing_id": raw.get("ListingId") or raw.get("ListingKey"), + "address": raw.get("UnparsedAddress") + or raw.get("StreetNumber", "") + " " + raw.get("StreetName", ""), + "city": raw.get("City"), + "state": raw.get("StateOrProvince"), + "zip_code": raw.get("PostalCode"), + "price": Decimal(str(raw.get("ListPrice", 0))) + if raw.get("ListPrice") + else None, + "beds": int(raw.get("BedroomsTotal", 0)) if raw.get("BedroomsTotal") else None, + "baths": Decimal(str(raw.get("BathroomsTotalInteger", 0))) + if raw.get("BathroomsTotalInteger") + else None, + "sq_ft": int(raw.get("LivingArea", 0)) if raw.get("LivingArea") else None, + "lot_size_sqft": int(raw.get("LotSizeSquareFeet", 0)) + if raw.get("LotSizeSquareFeet") + else None, + "property_type": normalize_property_type(raw.get("PropertyType")), + "property_sub_type": raw.get("PropertySubType"), + "year_built": int(raw.get("YearBuilt", 0)) if raw.get("YearBuilt") else None, + "lot_size_acres": Decimal(str(raw.get("LotSizeAcres", 0))) + if raw.get("LotSizeAcres") + else None, + "days_on_market": int(raw.get("DaysOnMarket", 0)) + if raw.get("DaysOnMarket") + else None, + "listing_status": raw.get("StandardStatus"), + "listing_date": raw.get("ListingContractDate"), + "expiration_date": raw.get("ExpirationDate"), + "mls_number": raw.get("MlsNumber") or raw.get("MlsId"), + "mls_id": raw.get("MlsId"), + "latitude": Decimal(str(raw.get("Latitude", 0))) + if raw.get("Latitude") + else None, + "longitude": Decimal(str(raw.get("Longitude", 0))) + if raw.get("Longitude") + else None, + "photos": [ + { + "url": m.get("MediaURL"), + "type": m.get("MediaType", "Photo"), + "caption": m.get("Description"), + "order": m.get("Order", 0), + } + for m in raw.get("Media", []) + if m.get("MediaURL") + ], + "virtual_tour_url": raw.get("VirtualTourURL"), + "listing_url": raw.get("PublicRemarks") or raw.get("ListingURL"), + "remarks": raw.get("PublicRemarks"), + "private_remarks": raw.get("PrivateRemarks"), + "agent_id": raw.get("ListAgentKey") or raw.get("ListAgentMlsId"), + "office_id": raw.get("ListOfficeKey") or raw.get("ListOfficeMlsId"), + "raw_data": raw, + } + + +# Export main classes and functions +__all__ = [ + "RESOAdapter", + "RESOAPIError", + "RESOAuthenticationError", + "RESORateLimitError", + "RESOAPIError", + "normalize_property_type", + "normalize_property_data", +] diff --git a/core/management/commands/update_market_indicators.py b/core/management/commands/update_market_indicators.py new file mode 100644 index 00000000..c983e20f --- /dev/null +++ b/core/management/commands/update_market_indicators.py @@ -0,0 +1,43 @@ +"""Management command to update market indicators from external data sources.""" + +from django.core.management.base import BaseCommand + +from core.integrations.market.market_trends import update_market_indicators + + +class Command(BaseCommand): + """Update market indicators from external data sources. + + Usage: + python manage.py update_market_indicators [--metro METRO_AREA] + + Examples: + python manage.py update_market_indicators + python manage.py update_market_indicators --metro "Dallas-Fort Worth-Arlington, TX" + """ + + help = "Update market cycle indicators from external data sources" + + def add_arguments(self, parser): + parser.add_argument( + "--metro", + type=str, + help="Specific metro area to update (e.g., 'Dallas-Fort Worth-Arlington, TX'). If not provided, updates all tracked metros.", + ) + + def handle(self, *args, **options): + metro: str = options.get("metro") or "" + if metro: + self.stdout.write(f"Updating market indicators for {metro}...") + else: + self.stdout.write("Updating market indicators for all tracked metros...") + + result = update_market_indicators(metro) + + self.stdout.write( + self.style.SUCCESS( + f"Updated market indicators: {result['created']} created, " + f"{result['updated']} updated, {result['errors']} errors" + ) + ) + self.stdout.write(f"Metro areas processed: {', '.join(result['metro_areas'])}") diff --git a/core/migrations/0050_marketindicator.py b/core/migrations/0050_marketindicator.py new file mode 100644 index 00000000..784f11ea --- /dev/null +++ b/core/migrations/0050_marketindicator.py @@ -0,0 +1,96 @@ +# Generated by Django 6.0.7 on 2026-08-21 16:48 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0049_merge_0048_capexitem_0048_financingscenario"), + ] + + operations = [ + migrations.CreateModel( + name="MarketIndicator", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "metro_area", + models.CharField( + db_index=True, + help_text="Metropolitan Statistical Area name (e.g., 'Dallas-Fort Worth-Arlington, TX')", + max_length=255, + ), + ), + ( + "indicator_type", + models.CharField( + choices=[ + ("median_price", "Median Home Price"), + ("dom", "Days on Market"), + ("months_supply", "Months of Supply"), + ("price_to_income", "Price-to-Income Ratio"), + ("rent_growth_yoy", "Rent Growth Year-over-Year"), + ], + db_index=True, + max_length=32, + ), + ), + ( + "value", + models.DecimalField( + decimal_places=4, + help_text="Indicator value (e.g., 425000.00 for median price, 28.0000 for DOM)", + max_digits=18, + ), + ), + ( + "date_recorded", + models.DateField( + db_index=True, + help_text="Date this indicator value was recorded", + ), + ), + ( + "source", + models.CharField( + blank=True, + default="", + help_text="Data source (e.g., 'zillow', 'census', 'bls', 'fred', 'manual')", + max_length=64, + ), + ), + ( + "notes", + models.TextField( + blank=True, + default="", + help_text="Additional context or methodology notes", + ), + ), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("updated_at", models.DateTimeField(auto_now=True)), + ], + options={ + "ordering": ["-date_recorded", "metro_area", "indicator_type"], + "indexes": [ + models.Index( + fields=["metro_area", "date_recorded"], + name="core_market_metro_a_045f3e_idx", + ), + models.Index( + fields=["indicator_type", "date_recorded"], + name="core_market_indicat_d192ce_idx", + ), + ], + "unique_together": {("metro_area", "indicator_type", "date_recorded")}, + }, + ), + ] diff --git a/core/models/growth.py b/core/models/growth.py index 62e8f7c7..b715d7fd 100644 --- a/core/models/growth.py +++ b/core/models/growth.py @@ -7,6 +7,93 @@ User = get_user_model() +class MarketIndicatorType(models.TextChoices): + """Market cycle indicator types for tracking real estate market health.""" + + MEDIAN_PRICE = "median_price", "Median Home Price" + DOM = "dom", "Days on Market" + MONTHS_SUPPLY = "months_supply", "Months of Supply" + PRICE_TO_INCOME = "price_to_income", "Price-to-Income Ratio" + RENT_GROWTH_YOY = "rent_growth_yoy", "Rent Growth Year-over-Year" + + +class MarketIndicator(models.Model): + """Market cycle indicator for tracking real estate market health metrics. + + Stores time-series data for key market indicators per metropolitan area. + Used to assess market cycle phase and health. + """ + + metro_area = models.CharField( + max_length=255, + db_index=True, + help_text="Metropolitan Statistical Area name (e.g., 'Dallas-Fort Worth-Arlington, TX')", + ) + indicator_type = models.CharField( + max_length=32, + choices=MarketIndicatorType.choices, + db_index=True, + ) + value = models.DecimalField( + max_digits=18, + decimal_places=4, + help_text="Indicator value (e.g., 425000.00 for median price, 28.0000 for DOM)", + ) + date_recorded = models.DateField( + db_index=True, + help_text="Date this indicator value was recorded", + ) + source = models.CharField( + max_length=64, + blank=True, + default="", + help_text="Data source (e.g., 'zillow', 'census', 'bls', 'fred', 'manual')", + ) + notes = models.TextField( + blank=True, + default="", + help_text="Additional context or methodology notes", + ) + + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + class Meta: + ordering = ["-date_recorded", "metro_area", "indicator_type"] + unique_together = ["metro_area", "indicator_type", "date_recorded"] + indexes = [ + models.Index(fields=["metro_area", "date_recorded"]), + models.Index(fields=["indicator_type", "date_recorded"]), + ] + + def __str__(self) -> str: # noqa: D401 + return f"{self.metro_area} - {self.get_indicator_type_display()} = {self.value} ({self.date_recorded})" + + def health_status(self) -> str: + """Determine market health status for this indicator. + + Returns: 'healthy', 'caution', or 'overheated' + """ + from core.integrations.market.market_trends import classify_market_health + + median_income = None + # Try to get median income from market snapshot if available + from core.models.growth import MarketSnapshot + + snapshot = MarketSnapshot.objects.filter( + msa_name__icontains=self.metro_area.split(",")[0].strip() + ).first() + if snapshot and snapshot.median_household_income: + median_income = snapshot.median_household_income + + return classify_market_health( + indicator_type=self.indicator_type, + value=self.value, + metro_area=self.metro_area, + median_income=median_income, + ) + + def compute_net_migration( population: int | None, pop_growth_rate: Decimal | None, diff --git a/core/services/saved_search_notifications.py b/core/services/saved_search_notifications.py new file mode 100644 index 00000000..fe22fa22 --- /dev/null +++ b/core/services/saved_search_notifications.py @@ -0,0 +1,134 @@ +"""Saved Search Notifications Service. + +Checks new listings against user saved searches and creates notifications +when matching properties are found. +""" + +from __future__ import annotations + +import logging +from datetime import timedelta +from decimal import Decimal +from typing import Any, Dict, List + +from django.utils import timezone + +from core.models import SavedSearch + +logger = logging.getLogger(__name__) + + +def get_saved_search_matches( + saved_search: SavedSearch, + listings: List[Dict[str, Any]], +) -> List[Dict[str, Any]]: + """Get listings that match a saved search criteria. + + Args: + saved_search: SavedSearch model instance with filter criteria + listings: List of listing dictionaries with price, state, zip_code + + Returns: + List of matching listings + """ + matches = [] + + for listing in listings: + # Check price range + price = listing.get("price") + if saved_search.min_price and price and price < saved_search.min_price: + continue + if saved_search.max_price and price and price > saved_search.max_price: + continue + + # Check state + if saved_search.state and listing.get("state") != saved_search.state: + continue + + # Check zip code + if saved_search.zip_code and listing.get("zip_code") != saved_search.zip_code: + continue + + matches.append(listing) + + return matches + + +def create_notification_for_match( + saved_search: SavedSearch, + listing: Dict[str, Any], +) -> Any: + """Create a notification for a saved search match. + + Args: + saved_search: The saved search that found a match + listing: The listing that matched + + Returns: + Created notification object or None + """ + from core.models import Notification + + # Create in-app notification + notification = Notification.objects.create( + user=saved_search.user, + notification_type="saved_search_match", + title=f"New listing matches '{saved_search.name}'", + body=f"{listing.get('address', 'Unknown address')} - ${listing.get('price', 0)}", + data={ + "saved_search_id": saved_search.id, + "listing": { + k: str(v) if isinstance(v, Decimal) else v for k, v in listing.items() + }, + }, + ) + + return notification + + +def check_listings_against_saved_searches() -> Dict[str, Any]: + """Check new listings against all active saved searches. + + Returns: + Dictionary with statistics about the check + """ + # Get all active saved searches + saved_searches = SavedSearch.objects.filter( + created_at__gte=timezone.now() - timedelta(days=30) # Last 30 days + ) + + stats = { + "searches_checked": 0, + "matches_found": 0, + "notifications_created": 0, + } + + # Fetch new listings (this would normally call an external API) + listings = fetch_new_listings() + + for saved_search in saved_searches: + matches = get_saved_search_matches(saved_search, listings) + if matches: + for match in matches: + notification = create_notification_for_match(saved_search, match) + if notification: + stats["notifications_created"] += 1 + stats["matches_found"] += len(matches) + + stats["searches_checked"] += 1 + + return stats + + +def fetch_new_listings() -> List[Dict[str, Any]]: + """Fetch new listings from external data sources. + + In production, this would call the RESO Web API, ATTOM, or other + data sources to get new listings. For now, returns empty list. + + Returns: + List of new listing dictionaries + """ + # Placeholder implementation - in production would call RESO Web API + # or other data sources + return [] diff --git a/core/urls.py b/core/urls.py index b414531e..1f913480 100644 --- a/core/urls.py +++ b/core/urls.py @@ -23,6 +23,12 @@ "properties//export/pdf/", views.export_pdf, name="property_export_pdf" ), path("properties//", views.property_detail, name="property_detail"), + path("markets/", views.market_dashboard, name="markets_dashboard"), + path( + "markets/update/", + views.update_market_indicators, + name="update_market_indicators", + ), path( "properties//financing/", views.financing_comparison, diff --git a/core/views/__init__.py b/core/views/__init__.py index 63e6200a..fafa9c45 100644 --- a/core/views/__init__.py +++ b/core/views/__init__.py @@ -58,6 +58,7 @@ # Moved from deprecated investor_app.finance.utils: from core.services.scoring import score_listing from core.services.financing_comparison import compare_scenarios, get_best_scenario +from core.integrations.market.market_trends import get_market_health_summary # keep only the models that are actually used from core.services.cma import estimate_listing_kpis, find_undervalued, price_per_sqft @@ -1536,6 +1537,57 @@ def portfolio_dashboard(request: HttpRequest) -> HttpResponse: ) +@login_required +def market_dashboard(request: HttpRequest) -> HttpResponse: + """Market cycle indicators dashboard — shows key market health metrics by metro area.""" + from core.models.growth import MarketIndicator + + metro_filter = request.GET.get("metro", "").strip() + + metro_qs = ( + MarketIndicator.objects.values("metro_area").distinct().order_by("metro_area") + ) + if metro_filter: + metro_qs = metro_qs.filter(metro_area__icontains=metro_filter) + + market_data = [] + for metro in metro_qs: + metro_name = metro["metro_area"] + summary = get_market_health_summary(metro_name) + indicators = summary.get("indicators", {}) + + market_data.append( + { + "metro_area": metro_name, + "overall_health": summary.get("overall_health", "caution"), + "health_counts": summary.get("health_counts", {}), + "indicators": indicators, + } + ) + + return render( + request, + "markets/dashboard.html", + { + "market_data": market_data, + "metro_filter": metro_filter, + }, + ) + + +@login_required +def update_market_indicators(request: HttpRequest) -> HttpResponse: + """Update market indicators from external data sources.""" + from core.integrations.market.market_trends import update_market_indicators + + result = update_market_indicators() + messages.success( + request, + f"Updated {result['created']} new indicators, {result['updated']} updated, {result['errors']} errors", + ) + return redirect("markets_dashboard") + + @login_required def pipeline_list(request: HttpRequest) -> HttpResponse: """Pipeline property list with stage funnel and filtering. diff --git a/investor_app/finance/tax_strategy.py b/investor_app/finance/tax_strategy.py new file mode 100644 index 00000000..cc870750 --- /dev/null +++ b/investor_app/finance/tax_strategy.py @@ -0,0 +1,295 @@ +"""Tax Strategy Module for Real Estate Investment Analysis. + +Implements: +1. QBI (Qualified Business Income) deduction - 20% deduction for qualified RE income +2. Passive Activity Loss (PAL) rules - $25k allowance phased out $100k-$150k AGI +3. 1031 Exchange analysis - defer capital gains by reinvesting in like-kind property + +References: +- IRC Section 199A - QBI Deduction +- IRC Section 469 - Passive Activity Loss Rules +- IRC Section 1031 - Like-Kind Exchanges +""" + +from __future__ import annotations + +from decimal import Decimal + +from investor_app.finance.utils import to_decimal + +# ── Constants ───────────────────────────────────────────────────────────────── + +QBI_DEDUCTION_RATE = Decimal("0.20") # 20% QBI deduction rate +PAL_FULL_ALLOWANCE = Decimal("25000") # Maximum $25k PAL deduction +PAL_PHASEOUT_START = Decimal("100000") # AGI threshold for phase-out +PAL_PHASEOUT_END = Decimal("150000") # AGI threshold for full phase-out + + +# ── QBI Deduction ───────────────────────────────────────────────────────────── + + +def calculate_qbi_deduction( + qualified_business_income: Decimal, + w2_wages: Decimal, + qbi_adjusted_basis: Decimal, + taxable_income: Decimal = Decimal("0"), +) -> Decimal: + """Calculate Qualified Business Income (QBI) deduction under IRC Section 199A. + + The QBI deduction is generally 20% of qualified business income, subject to + limitations for high-income taxpayers (W-2 wage limit, property basis limit). + + Args: + qualified_business_income: Net qualified business income from the property. + w2_wages: Total W-2 wages paid by the business (for wage limitation). + qbi_adjusted_basis: Unadjusted basis of qualified property (for basis limitation). + taxable_income: Taxable income for determining phase-out (default 0). + + Returns: + QBI deduction amount as Decimal. + """ + qbi = to_decimal(qualified_business_income) + wages = to_decimal(w2_wages) + basis = to_decimal(qbi_adjusted_basis) + income = to_decimal(taxable_income) + + # Base QBI deduction: 20% of qualified business income + base_deduction = qbi * QBI_DEDUCTION_RATE + + # For high-income taxpayers, apply W-2 wage and property limitations + # Phase-out begins at $164,900 for single filers (2024) + # Simplified: if taxable income is above threshold, apply limitations + phaseout_threshold = Decimal("164900") + + if income > phaseout_threshold: + # W-2 wage limitation: 50% of W-2 wages + wage_limit = wages * Decimal("0.50") + + # Qualified property limitation: 2.5% of QBI adjusted basis + property_limit = basis * Decimal("0.025") + + # Use the greater of wage limit or property limit + limitation = max(wage_limit, property_limit) + + # QBI deduction is limited to the lesser of base deduction or limitation + return min(base_deduction, limitation).quantize(Decimal("0.01")) + + # Below phase-out: full 20% deduction + return base_deduction.quantize(Decimal("0.01")) + + +# ── Passive Activity Loss (PAL) Rules ──────────────────────────────────────── + + +def calculate_pal_phase_out(modified_agi: Decimal) -> Decimal: + """Calculate PAL phase-out percentage based on Modified AGI. + + The $25,000 rental loss allowance phases out linearly between + $100,000 and $150,000 of Modified Adjusted Gross Income (MAGI). + + Args: + modified_agi: Modified Adjusted Gross Income. + + Returns: + Phase-out percentage as Decimal (1.0 = no reduction, 0.0 = fully phased out) + """ + magi = to_decimal(modified_agi) + + if magi <= PAL_PHASEOUT_START: + return Decimal("1") + elif magi >= PAL_PHASEOUT_END: + return Decimal("0") + else: + # Linear phase-out: (150000 - magi) / 50000 + reduction = (PAL_PHASEOUT_END - magi) / (PAL_PHASEOUT_END - PAL_PHASEOUT_START) + return reduction.quantize(Decimal("0.01")) + + +def calculate_pal_allowance( + active_participation: Decimal, + modified_agi: Decimal, + rental_losses: Decimal, +) -> Decimal: + """Calculate Passive Activity Loss (PAL) deduction allowance. + + Under IRC Section 469, rental real estate losses can offset up to $25,000 + of other income if the taxpayer actively participates, with phase-out + for incomes between $100k-$150k MAGI. + + Args: + active_participation: 1.0 if active participation, 0.0 otherwise. + modified_agi: Modified Adjusted Gross Income. + rental_losses: Passive rental losses for the year. + + Returns: + PAL deduction amount (capped at $25,000 and actual losses). + """ + participation = to_decimal(active_participation) + losses = to_decimal(rental_losses) + + # No deduction without active participation + if participation <= 0: + return Decimal("0") + + # Calculate phase-out adjusted allowance + phase_out = calculate_pal_phase_out(to_decimal(modified_agi)) + allowance = PAL_FULL_ALLOWANCE * phase_out + + # Capped at actual losses + return min(allowance, losses).quantize(Decimal("0.01")) + + +# ── 1031 Exchange Analysis ──────────────────────────────────────────────────── + + +def calculate_1031_deferral_ratio( + sale_price: Decimal, + replacement_price: Decimal, + selling_costs: Decimal, +) -> Decimal: + """Calculate the ratio of gain that can be deferred in a 1031 exchange. + + Deferral ratio = replacement_price / (sale_price - selling_costs) + Capped at 1.0 when replacement price >= net sale price (100% deferral). + + Args: + sale_price: Gross sale price of relinquished property. + replacement_price: Purchase price of replacement property. + selling_costs: Costs to sell (agent commission, closing costs). + + Returns: + Deferral ratio as Decimal (1.0 = 100% deferral, <1.0 = partial deferral). + """ + sp = to_decimal(sale_price) + rp = to_decimal(replacement_price) + sc = to_decimal(selling_costs) + + net_proceeds = sp - sc + if net_proceeds <= 0: + return Decimal("0") + + ratio = rp / net_proceeds + + # Cap at 1.0 for 100% deferral + return min(ratio, Decimal("1")).quantize(Decimal("0.0001")) + + +def calculate_1031_exchange( + sale_price: Decimal, + original_cost_basis: Decimal, + accumulated_depreciation: Decimal, + replacement_price: Decimal, + selling_costs: Decimal, +) -> dict[str, Decimal]: + """Calculate 1031 exchange outcomes and deferral. + + In a 1031 exchange, capital gains tax is deferred by reinvesting + sale proceeds into a like-kind property of equal or greater value. + + Args: + sale_price: Gross sale price of relinquished property. + original_cost_basis: Original purchase price of relinquished property. + accumulated_depreciation: Total depreciation taken. + replacement_price: Purchase price of replacement property. + selling_costs: Costs to sell (agent commission, closing costs). + + Returns: + Dictionary with exchange metrics: + - deferred_gain: Portion of gain deferred + - boot_received: Taxable boot (if replacement price < net sale price) + - adjusted_basis: New basis in replacement property + - depreciation_recapture: Depreciation recapture on boot + """ + sp = to_decimal(sale_price) + cb = to_decimal(original_cost_basis) + ad = to_decimal(accumulated_depreciation) + rp = to_decimal(replacement_price) + sc = to_decimal(selling_costs) + + # Calculate net sale proceeds + net_proceeds = sp - sc + + # Calculate realized gain + realized_gain = sp - cb + + # Calculate depreciation recapture (for reporting purposes) + if realized_gain > ad: + depreciation_recapture = ad + else: + depreciation_recapture = realized_gain + + # In a 1031 exchange, depreciation recapture is DEFERRED (not immediately taxed) + # Calculate boot (taxable portion if replacement price < net proceeds) + if rp >= net_proceeds: + # Full deferral - no boot, entire realized gain is deferred + boot_received = Decimal("0") + deferred_gain = realized_gain # Full gain deferred + else: + # Partial deferral - boot = net_proceeds - replacement_price + boot_received = net_proceeds - rp + deferred_gain = realized_gain - boot_received + + # Ensure deferred_gain is not negative + deferred_gain = max(deferred_gain, Decimal("0")) + + # Calculate adjusted basis for replacement property + adjusted_basis = rp + deferred_gain + + return { + "deferred_gain": deferred_gain.quantize(Decimal("0.01")), + "boot_received": boot_received.quantize(Decimal("0.01")), + "adjusted_basis": adjusted_basis.quantize(Decimal("0.01")), + "depreciation_recapture": depreciation_recapture.quantize(Decimal("0.01")), + } + + +# ── Combined Tax Benefit Calculator ──────────────────────────────────────────── + + +def calculate_total_tax_benefit( + qbi_income: Decimal, + rental_losses: Decimal, + modified_agi: Decimal, + marginal_tax_rate: Decimal, +) -> Decimal: + """Calculate total annual tax benefit from QBI and PAL. + + Combines QBI deduction and Passive Activity Loss allowance into + a single tax benefit amount. + + Args: + qbi_income: Net qualified business income from real estate activities. + rental_losses: Passive rental losses for the year. + modified_agi: Modified Adjusted Gross Income. + marginal_tax_rate: Marginal income tax rate as decimal (e.g., 0.24 for 24%). + + Returns: + Total tax benefit as Decimal. + """ + qbi = to_decimal(qbi_income) + losses = to_decimal(rental_losses) + agi = to_decimal(modified_agi) + rate = to_decimal(marginal_tax_rate) + + # Calculate QBI deduction (assume $0 W-2 wages and $0 basis for simplicity) + qbi_deduction = calculate_qbi_deduction( + qualified_business_income=qbi, + w2_wages=Decimal("0"), + qbi_adjusted_basis=Decimal("0"), + taxable_income=agi, + ) + + # Calculate PAL allowance + pal_allowance = calculate_pal_allowance( + active_participation=Decimal("1"), # Assume active participation + modified_agi=agi, + rental_losses=losses, + ) + + # Total deductions + total_deductions = qbi_deduction + pal_allowance + + # Tax benefit + tax_benefit = total_deductions * rate + + return tax_benefit.quantize(Decimal("0.01")) diff --git a/lighthouserc.json b/lighthouserc.json index 749ef3a5..c9280dfa 100644 --- a/lighthouserc.json +++ b/lighthouserc.json @@ -1,7 +1,12 @@ { "ci": { "collect": { - "numberOfRuns": 2 + "numberOfRuns": 2, + "staticDistDir": "./staticfiles", + "url": "http://localhost:8000/", + "startServerCommand": null, + "startServerTimeout": 120, + "maxWaitForLoad": 60000 }, "assert": { "preset": "lighthouse:no-pwa", diff --git a/static/icons/icon-128x128.svg b/static/icons/icon-128x128.svg new file mode 100644 index 00000000..fa6a4d36 --- /dev/null +++ b/static/icons/icon-128x128.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/icons/icon-144x144.svg b/static/icons/icon-144x144.svg new file mode 100644 index 00000000..564c7cf1 --- /dev/null +++ b/static/icons/icon-144x144.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/icons/icon-152x152.svg b/static/icons/icon-152x152.svg new file mode 100644 index 00000000..a36381e9 --- /dev/null +++ b/static/icons/icon-152x152.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/icons/icon-192x192.svg b/static/icons/icon-192x192.svg new file mode 100644 index 00000000..1de10f97 --- /dev/null +++ b/static/icons/icon-192x192.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/icons/icon-384x384.svg b/static/icons/icon-384x384.svg new file mode 100644 index 00000000..b01f387e --- /dev/null +++ b/static/icons/icon-384x384.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/icons/icon-512x512.svg b/static/icons/icon-512x512.svg new file mode 100644 index 00000000..a61374e8 --- /dev/null +++ b/static/icons/icon-512x512.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/icons/icon-72x72.svg b/static/icons/icon-72x72.svg new file mode 100644 index 00000000..06e9e403 --- /dev/null +++ b/static/icons/icon-72x72.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/icons/icon-96x96.svg b/static/icons/icon-96x96.svg new file mode 100644 index 00000000..c4cdd9ba --- /dev/null +++ b/static/icons/icon-96x96.svg @@ -0,0 +1,3 @@ + + + diff --git a/static/manifest.json b/static/manifest.json new file mode 100644 index 00000000..f03b6d8e --- /dev/null +++ b/static/manifest.json @@ -0,0 +1,87 @@ +{ + "name": "PREI - Real Estate Investment Analytics", + "short_name": "PREI", + "description": "Passive real estate investment analytics for buy-and-hold investors", + "start_url": "/", + "display": "standalone", + "background_color": "#F8F8F6", + "theme_color": "#1D9E75", + "orientation": "portrait-primary", + "scope": "/", + "icons": [ + { + "src": "/static/icons/icon-72x72.svg", + "sizes": "72x72", + "type": "image/svg+xml", + "purpose": "any maskable" + }, + { + "src": "/static/icons/icon-96x96.svg", + "sizes": "96x96", + "type": "image/svg+xml", + "purpose": "any maskable" + }, + { + "src": "/static/icons/icon-128x128.svg", + "sizes": "128x128", + "type": "image/svg+xml", + "purpose": "any maskable" + }, + { + "src": "/static/icons/icon-144x144.svg", + "sizes": "144x144", + "type": "image/svg+xml", + "purpose": "any maskable" + }, + { + "src": "/static/icons/icon-152x152.svg", + "sizes": "152x152", + "type": "image/svg+xml", + "purpose": "any maskable" + }, + { + "src": "/static/icons/icon-192x192.svg", + "sizes": "192x192", + "type": "image/svg+xml", + "purpose": "any maskable" + }, + { + "src": "/static/icons/icon-384x384.svg", + "sizes": "384x384", + "type": "image/svg+xml", + "purpose": "any maskable" + }, + { + "src": "/static/icons/icon-512x512.svg", + "sizes": "512x512", + "type": "image/svg+xml", + "purpose": "any maskable" + } + ], + "categories": ["business", "finance", "productivity"], + "shortcuts": [ + { + "name": "Dashboard", + "short_name": "Dashboard", + "description": "View your portfolio dashboard", + "url": "/dashboard/", + "icons": [{ "src": "/static/icons/icon-192x192.png", "sizes": "192x192" }] + }, + { + "name": "Growth Areas", + "short_name": "Growth", + "description": "Explore growth areas", + "url": "/growth/", + "icons": [{ "src": "/static/icons/icon-192x192.png", "sizes": "192x192" }] + }, + { + "name": "Pipeline", + "short_name": "Pipeline", + "description": "View your property pipeline", + "url": "/pipeline/", + "icons": [{ "src": "/static/icons/icon-192x192.png", "sizes": "192x192" }] + } + ], + "related_applications": [], + "prefer_related_applications": false +} diff --git a/static/sw.js b/static/sw.js new file mode 100644 index 00000000..bea652e5 --- /dev/null +++ b/static/sw.js @@ -0,0 +1,217 @@ +// Service Worker for PREI - Cache-first strategy for offline support +// Version: 1.0.0 + +const CACHE_NAME = 'prei-v1'; +const OFFLINE_URL = '/offline/'; + +// Static assets to cache on install +const STATIC_ASSETS = [ + '/', + '/offline/', + '/static/css/tokens.css', + '/static/css/base.css', + '/static/js/theme.js', + '/static/js/notifications.js', + '/static/manifest.json', +]; + +// Cache-first strategy for static assets +async function cacheFirst(request) { + const cache = await caches.open(CACHE_NAME); + const cachedResponse = await cache.match(request); + + if (cachedResponse) { + // Return cached response and update cache in background + const fetchPromise = fetch(request).then(response => { + if (response.ok) { + cache.put(request, response.clone()); + } + return response; + }).catch(() => cachedResponse); // Return cached if network fails + + return cachedResponse; + } + + // Not in cache, fetch and cache + try { + const networkResponse = await fetch(request); + if (networkResponse.ok) { + const cache = await caches.open(CACHE_NAME); + cache.put(request, networkResponse.clone()); + } + return networkResponse; + } catch (error) { + // If it's a navigation request and we're offline, show offline page + if (request.mode === 'navigate') { + const cache = await caches.open(CACHE_NAME); + const offlineResponse = await cache.match(OFFLINE_URL); + if (offlineResponse) { + return offlineResponse; + } + } + throw error; + } +} + +// Network-first strategy for API calls +async function networkFirst(request) { + const cache = await caches.open(CACHE_NAME); + + try { + const networkResponse = await fetch(request); + if (networkResponse.ok) { + cache.put(request, networkResponse.clone()); + } + return networkResponse; + } catch (error) { + const cachedResponse = await cache.match(request); + if (cachedResponse) { + return cachedResponse; + } + throw error; + } +} + +// Stale-while-revalidate for API responses +async function staleWhileRevalidate(request) { + const cache = await caches.open(CACHE_NAME); + const cachedResponse = await cache.match(request); + + const fetchPromise = fetch(request).then(async (networkResponse) => { + if (networkResponse.ok) { + cache.put(request, networkResponse.clone()); + } + return networkResponse; + }); + + if (cachedResponse) { + // Return cached immediately, update in background + fetchPromise.catch(() => {}); // suppress error + return cachedResponse; + } + + return fetchPromise; +} + +// Install event - cache static assets +self.addEventListener('install', event => { + event.waitUntil( + caches.open(CACHE_NAME) + .then(cache => cache.addAll(STATIC_ASSETS)) + .then(() => self.skipWaiting()) + ); +}); + +// Activate event - clean up old caches +self.addEventListener('activate', event => { + event.waitUntil( + caches.keys().then(cacheNames => { + return Promise.all( + cacheNames + .filter(name => name !== CACHE_NAME) + .map(name => caches.delete(name)) + ); + }).then(() => self.clients.claim()) + ); +}); + +// Fetch event - route requests to appropriate strategy +self.addEventListener('fetch', event => { + const { request } = event; + const url = new URL(request.url); + + // Skip non-GET requests + if (request.method !== 'GET') { + return; + } + + // Skip cross-origin requests + if (url.origin !== location.origin) { + return; + } + + // Handle different routes with appropriate strategies + if (url.pathname.startsWith('/static/')) { + // Static assets: cache-first + event.respondWith(cacheFirst(event.request)); + } else if (url.pathname.startsWith('/api/')) { + // API calls: stale-while-revalidate + event.respondWith(staleWhileRevalidate(event.request)); + } else if (request.mode === 'navigate') { + // Navigation requests: network-first with offline fallback + event.respondWith(networkFirst(event.request)); + } else { + // Default: stale-while-revalidate + event.respondWith(staleWhileRevalidate(event.request)); + } +}); + +// Background sync for offline form submissions +self.addEventListener('sync', event => { + if (event.tag === 'sync-forms') { + event.waitUntil(syncForms()); + } +}); + +async function syncForms() { + const cache = await caches.open('form-submissions'); + const requests = await cache.keys(); + + for (const request of requests) { + try { + await fetch(request); + await cache.delete(request); + } catch (error) { + console.error('Sync failed for', request.url, error); + } + } +} + +// Push notification handling +self.addEventListener('push', event => { + if (!event.data) return; + + const data = event.data.json(); + const options = { + body: data.body, + icon: '/static/icons/icon-192x192.png', + badge: '/static/icons/badge-72x72.png', + vibrate: [100, 50, 100], + data: { + url: data.url || '/' + } + }; + + event.waitUntil( + self.registration.showNotification(data.title, options) + ); +}); + +self.addEventListener('notificationclick', event => { + event.notification.close(); + + event.waitUntil( + clients.matchAll({ type: 'window', includeUncontrolled: true }) + .then(clientList => { + for (const client of clientList) { + if (client.url === event.notification.data.url && 'focus' in client) { + return client.focus(); + } + } + return clients.openWindow(event.notification.data.url); + }) + ); +}); + +// Periodic background sync (if supported) +self.addEventListener('periodicsync', event => { + if (event.tag === 'update-indicators') { + event.waitUntil(updateMarketIndicators()); + } +}); + +async function updateMarketIndicators() { + // This would trigger a background update of market indicators + // Implementation depends on your backend API + console.log('Periodic sync: updating market indicators'); +} diff --git a/templates/base.html b/templates/base.html index 29188d49..dfdb97c0 100644 --- a/templates/base.html +++ b/templates/base.html @@ -4,10 +4,22 @@ {% block title %}PREI{% endblock %} — Real Estate Investor + + {% load static %} + + + + + + + + + + - {% load static %} + {% block extra_css %}{% endblock %} @@ -137,5 +149,32 @@ {% block extra_js %}{% endblock %} + + + diff --git a/templates/growth_areas.html b/templates/growth_areas.html index a259de03..65702ca9 100644 --- a/templates/growth_areas.html +++ b/templates/growth_areas.html @@ -97,6 +97,8 @@

Growth Areas

class="btn btn-primary">Discover Properties View Screened + Market Indicators diff --git a/templates/markets/dashboard.html b/templates/markets/dashboard.html new file mode 100644 index 00000000..a8d0a87a --- /dev/null +++ b/templates/markets/dashboard.html @@ -0,0 +1,245 @@ +{% extends "base.html" %} +{% load humanize %} +{% block title %}Market Cycle Indicators — prei{% endblock %} + +{% block content %} +
+
+

Market Cycle Indicators

+

Track median price, DOM, supply, and rent growth across metro areas

+
+
+
+ {% csrf_token %} + +
+
+
+ +{% if market_data %} +
+

Market Overview

+
+ {% for market in market_data %} +
+
{{ market.metro_area }}
+
{{ market.overall_health|capfirst }}
+
+ Healthy {{ market.health_counts.healthy }} + Caution {{ market.health_counts.caution }} + Overheated {{ market.health_counts.overheated }} +
+
+ {% endfor %} +
+
+ +
+

Indicator Details

+ {% for market in market_data %} +
+

{{ market.metro_area }}

+
+ {% for ind_type, data in market.indicators.items %} +
+
+ {{ ind_type|capfirst }} + {{ data.health|capfirst }} +
+
+ {% if data.value >= 1000000 %} + ${{ data.value|floatformat:0|intcomma }} + {% elif data.value >= 1000 %} + ${{ data.value|floatformat:0|intcomma }} + {% elif data.value >= 1 %} + ${{ data.value|floatformat:1 }} + {% else %} + {{ data.value|floatformat:2 }}% + {% endif %} +
+ {% if data.sparkline %} + + + + {% else %} +
No history yet
+ {% endif %} +
+ Source: {{ data.source|default:"N/A" }} · {{ data.date_recorded }} +
+
+ {% endfor %} +
+
+ {% endfor %} +
+{% else %} +
+

No market indicator data available. Run python manage.py update_market_indicators to populate.

+
+{% endif %} +{% endblock %} + +{% block extra_css %} + +{% endblock %} diff --git a/templates/offline.html b/templates/offline.html new file mode 100644 index 00000000..2adbf742 --- /dev/null +++ b/templates/offline.html @@ -0,0 +1,155 @@ +{% extends "base.html" %} +{% load static %} + +{% block title %}Offline - PREI{% endblock %} + +{% block extra_css %} + +{% endblock %} + +{% block content %} +
+ + +

You're Offline

+ +

+ It looks like you've lost your internet connection. Don't worry — PREI works offline for previously loaded pages. +

+ +
+ + + Go Home +
+ +
+
Available Offline
+
+
+ + Cached pages +
+
+ + Cached calculations +
+
+ + Saved reports +
+
+
+
+{% endblock %} diff --git a/tests/test_market_indicators.py b/tests/test_market_indicators.py new file mode 100644 index 00000000..5701ff14 --- /dev/null +++ b/tests/test_market_indicators.py @@ -0,0 +1,390 @@ +"""Tests for market cycle indicators.""" + +from decimal import Decimal + +import pytest + +from core.models.growth import MarketIndicator, MarketIndicatorType + + +class TestMarketIndicatorModel: + """Tests for MarketIndicator model.""" + + @pytest.mark.django_db + def test_market_indicator_creation(self) -> None: + """Test creating a MarketIndicator with all required fields.""" + indicator = MarketIndicator.objects.create( + metro_area="Dallas-Fort Worth-Arlington, TX", + indicator_type=MarketIndicatorType.MEDIAN_PRICE, + value=Decimal("425000"), + date_recorded="2026-01-15", + ) + assert indicator.metro_area == "Dallas-Fort Worth-Arlington, TX" + assert indicator.indicator_type == MarketIndicatorType.MEDIAN_PRICE + assert indicator.value == Decimal("425000") + assert str(indicator.date_recorded) == "2026-01-15" + + @pytest.mark.django_db + def test_market_indicator_types(self) -> None: + """Test all indicator types can be created.""" + types = [ + MarketIndicatorType.MEDIAN_PRICE, + MarketIndicatorType.DOM, + MarketIndicatorType.MONTHS_SUPPLY, + MarketIndicatorType.PRICE_TO_INCOME, + MarketIndicatorType.RENT_GROWTH_YOY, + ] + for t in types: + indicator = MarketIndicator.objects.create( + metro_area="Test Metro", + indicator_type=t, + value=Decimal("100"), + date_recorded="2026-01-15", + ) + assert indicator.indicator_type == t + + @pytest.mark.django_db + def test_unique_constraint_metro_type_date(self) -> None: + """Test that metro_area + indicator_type + date_recorded is unique.""" + MarketIndicator.objects.create( + metro_area="Test Metro", + indicator_type=MarketIndicatorType.MEDIAN_PRICE, + value=Decimal("100"), + date_recorded="2026-01-15", + ) + # Creating another with same metro, type, date should fail + with pytest.raises(Exception): + MarketIndicator.objects.create( + metro_area="Test Metro", + indicator_type=MarketIndicatorType.MEDIAN_PRICE, + value=Decimal("200"), + date_recorded="2026-01-15", + ) + + +class TestMarketIndicatorClassification: + """Tests for market indicator classification/health scoring.""" + + @pytest.mark.django_db + def test_classify_median_price_healthy(self) -> None: + """Test median price classification - moderate price = healthy.""" + from core.integrations.market.market_trends import classify_market_health + + # Median price around 3x income = healthy + health = classify_market_health( + indicator_type="median_price", + value=Decimal("350000"), + metro_area="Test Metro", + median_income=Decimal("100000"), + ) + assert health == "healthy" + + @pytest.mark.django_db + def test_classify_median_price_overheated(self) -> None: + """Test median price classification - high price = overheated.""" + from core.integrations.market.market_trends import classify_market_health + + # Price > 5x income = overheated + health = classify_market_health( + indicator_type="median_price", + value=Decimal("600000"), + metro_area="Test Metro", + median_income=Decimal("100000"), + ) + assert health == "overheated" + + @pytest.mark.django_db + def test_classify_dom_healthy(self) -> None: + """Test DOM classification - moderate DOM = healthy.""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="dom", + value=Decimal("30"), + metro_area="Test Metro", + ) + assert health == "healthy" + + @pytest.mark.django_db + def test_classify_dom_caution(self) -> None: + """Test DOM classification - low DOM = caution (overheated).""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="dom", + value=Decimal("10"), + metro_area="Test Metro", + ) + assert health == "caution" + + @pytest.mark.django_db + def test_classify_months_supply_healthy(self) -> None: + """Test months supply classification - 4-6 months = healthy.""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="months_supply", + value=Decimal("5"), + metro_area="Test Metro", + ) + assert health == "healthy" + + @pytest.mark.django_db + def test_classify_months_supply_sellers_market(self) -> None: + """Test months supply classification - < 3 months = sellers market (overheated).""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="months_supply", + value=Decimal("2"), + metro_area="Test Metro", + ) + assert health == "overheated" + + @pytest.mark.django_db + def test_classify_price_to_income_healthy(self) -> None: + """Test price-to-income classification - 3-4 ratio = healthy.""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="price_to_income", + value=Decimal("3.5"), + metro_area="Test Metro", + ) + assert health == "healthy" + + @pytest.mark.django_db + def test_classify_rent_growth_healthy(self) -> None: + """Test rent growth classification - 3-5% = healthy.""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="rent_growth_yoy", + value=Decimal("0.04"), # 4% + metro_area="Test Metro", + ) + assert health == "healthy" + + @pytest.mark.django_db + def test_classify_rent_growth_overheated(self) -> None: + """Test rent growth classification - > 8% = overheated.""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="rent_growth_yoy", + value=Decimal("0.10"), # 10% + metro_area="Test Metro", + ) + assert health == "overheated" + + @pytest.mark.django_db + def test_classify_rent_growth_declining(self) -> None: + """Test rent growth classification - negative = declining (caution).""" + from core.integrations.market.market_trends import classify_market_health + + health = classify_market_health( + indicator_type="rent_growth_yoy", + value=Decimal("-0.02"), # -2% + metro_area="Test Metro", + ) + assert health == "caution" + + +class TestMarketTrendsAdapter: + """Tests for market trends adapter.""" + + @pytest.mark.django_db + def test_fetch_market_indicators_returns_data(self) -> None: + """Test that fetch_market_indicators returns indicator data.""" + from core.integrations.market.market_trends import fetch_market_indicators + + # This will test the adapter interface + # The actual implementation will use mock data or external APIs + result = fetch_market_indicators(metro_area="Test Metro") + assert isinstance(result, list) + # Each item should have indicator_type, value, date_recorded + for item in result: + assert "indicator_type" in item + assert "value" in item + assert "date_recorded" in item + + @pytest.mark.django_db + def test_get_latest_indicators(self) -> None: + """Test getting latest indicators for a metro area.""" + from core.integrations.market.market_trends import get_latest_indicators + + result = get_latest_indicators(metro_area="Test Metro") + assert isinstance(result, dict) + # Should have all 5 indicator types + expected_types = { + "median_price", + "dom", + "months_supply", + "price_to_income", + "rent_growth_yoy", + } + for t in expected_types: + assert t in result + + +class TestUpdateMarketIndicatorsCommand: + """Tests for update_market_indicators management command.""" + + @pytest.mark.django_db + def test_command_creates_indicators(self, capsys) -> None: + """Test that command creates market indicator records.""" + from django.core.management import call_command + + call_command("update_market_indicators", "--metro", "Test Metro") + captured = capsys.readouterr() + assert "Updated" in captured.out or "Created" in captured.out + + # Verify indicators were created + from core.models.growth import MarketIndicator + + indicators = MarketIndicator.objects.filter(metro_area="Test Metro") + assert indicators.count() >= 1 + + +class TestMarketDashboardView: + """Tests for market dashboard view.""" + + @pytest.mark.django_db + def test_dashboard_renders(self, client, user) -> None: + """Test dashboard renders successfully.""" + client.force_login(user) + response = client.get("/markets/") + assert response.status_code == 200 + + @pytest.mark.django_db + def test_dashboard_shows_indicators(self, client, user) -> None: + """Test that dashboard shows indicator cards.""" + from core.models.growth import MarketIndicator, MarketIndicatorType + + # Create some test data + MarketIndicator.objects.create( + metro_area="Dallas-Fort Worth-Arlington, TX", + indicator_type=MarketIndicatorType.MEDIAN_PRICE, + value=Decimal("425000"), + date_recorded="2026-01-15", + ) + + client.force_login(user) + response = client.get("/markets/") + assert response.status_code == 200 + assert "Dallas" in response.content.decode() + + +class TestSparklineHelpers: + """Tests for trend chart helpers.""" + + def test_build_sparkline_points_normalizes_values(self) -> None: + """Points should be normalized into the 100x30 viewBox.""" + from core.integrations.market.market_trends import build_sparkline_points + + points = build_sparkline_points([Decimal("0"), Decimal("5"), Decimal("10")]) + # Rising series: min at bottom-left, max at top-right + assert points == "0,30 50,15 100,0" + + def test_build_sparkline_points_flat_series(self) -> None: + """All-equal values should draw a flat line at the midpoint.""" + from core.integrations.market.market_trends import build_sparkline_points + + points = build_sparkline_points([Decimal("3"), Decimal("3"), Decimal("3")]) + assert points == "0,15 50,15 100,15" + + def test_build_sparkline_points_empty(self) -> None: + """Empty series produces no points.""" + from core.integrations.market.market_trends import build_sparkline_points + + assert build_sparkline_points([]) == "" + + @pytest.mark.django_db + def test_get_indicator_history_returns_oldest_first(self) -> None: + """History should be limited to recent values, oldest first.""" + from core.integrations.market.market_trends import get_indicator_history + from core.models.growth import MarketIndicator, MarketIndicatorType + + MarketIndicator.objects.create( + metro_area="History Metro", + indicator_type=MarketIndicatorType.DOM, + value=Decimal("40"), + date_recorded="2026-01-01", + ) + MarketIndicator.objects.create( + metro_area="History Metro", + indicator_type=MarketIndicatorType.DOM, + value=Decimal("20"), + date_recorded="2026-02-01", + ) + + history = get_indicator_history("History Metro", "dom") + assert history == [Decimal("40"), Decimal("20")] + + @pytest.mark.django_db + def test_get_indicator_history_respects_limit(self) -> None: + """Only the ``limit`` most recent values are returned.""" + from core.integrations.market.market_trends import get_indicator_history + from core.models.growth import MarketIndicator, MarketIndicatorType + + for month in range(1, 6): + MarketIndicator.objects.create( + metro_area="Limit Metro", + indicator_type=MarketIndicatorType.DOM, + value=Decimal(str(month)), + date_recorded=f"2025-{month:02d}-01", + ) + + history = get_indicator_history("Limit Metro", "dom", limit=3) + assert history == [Decimal("3"), Decimal("4"), Decimal("5")] + + @pytest.mark.django_db + def test_get_latest_indicators_prefers_db_records(self) -> None: + """Stored DB values should take precedence over adapter fallback.""" + from core.integrations.market.market_trends import get_latest_indicators + from core.models.growth import MarketIndicator, MarketIndicatorType + + MarketIndicator.objects.create( + metro_area="DB Metro", + indicator_type=MarketIndicatorType.MEDIAN_PRICE, + value=Decimal("999000"), + date_recorded="2026-01-15", + ) + + result = get_latest_indicators("DB Metro") + assert result["median_price"]["value"] == Decimal("999000") + assert result["median_price"]["history"] == [Decimal("999000")] + assert result["median_price"]["sparkline"] != "" + + +class TestMarketIndicatorsE2E: + """End-to-end tests for market indicators feature.""" + + @pytest.mark.django_db + def test_market_dashboard_renders_with_indicators(self, client, user) -> None: + """Test that market dashboard renders with indicators for tracked markets.""" + from core.models.growth import MarketIndicator, MarketIndicatorType + + # Create test data + MarketIndicator.objects.create( + metro_area="Dallas-Fort Worth-Arlington, TX", + indicator_type=MarketIndicatorType.MEDIAN_PRICE, + value=Decimal("425000"), + date_recorded="2026-01-15", + ) + MarketIndicator.objects.create( + metro_area="Dallas-Fort Worth-Arlington, TX", + indicator_type=MarketIndicatorType.DOM, + value=Decimal("28"), + date_recorded="2026-01-15", + ) + + client.force_login(user) + response = client.get("/markets/") + assert response.status_code == 200 + content = response.content.decode() + assert "Dallas" in content + assert "425000" in content or "425" in content # Check median price shown + assert "28" in content # Check DOM shown + # Trend chart (sparkline SVG) should render for each indicator + assert " 0 + + for icon in manifest["icons"]: + assert "src" in icon + assert "sizes" in icon + assert "type" in icon + + +class TestServiceWorker: + """Tests for service worker.""" + + def test_sw_exists(self): + """Test that sw.js exists in static directory.""" + sw_path = Path("static/sw.js") + assert sw_path.exists(), "sw.js should exist in static/" + + def test_sw_registers_in_base(self): + """Test that base.html registers the service worker.""" + base_path = Path("templates/base.html") + with open(base_path) as f: + content = f.read() + assert ( + "navigator.serviceWorker.register" in content + or "serviceWorker.register" in content + ) + + def test_sw_cache_strategy(self): + """Test that service worker implements cache-first strategy.""" + sw_path = Path("static/sw.js") + with open(sw_path) as f: + content = f.read() + + # Check for cache-first strategy patterns + assert "cache" in content.lower() or "cacheFirst" in content + assert "fetch" in content + + +class TestOfflineFallback: + """Tests for offline fallback page.""" + + def test_offline_page_exists(self): + """Test that offline.html exists.""" + offline_path = Path("templates/offline.html") + assert offline_path.exists(), "offline.html should exist in templates/" + + def test_offline_page_content(self): + """Test that offline page has appropriate content.""" + offline_path = Path("templates/offline.html") + with open(offline_path) as f: + content = f.read() + + assert "offline" in content.lower() + assert "internet" in content.lower() or "connection" in content.lower() + + +class TestBaseTemplate: + """Tests for base.html PWA integration.""" + + def test_manifest_link(self): + """Test that base.html links to manifest.json.""" + base_path = Path("templates/base.html") + with open(base_path) as f: + content = f.read() + assert 'rel="manifest"' in content + assert 'href="/static/manifest.json"' in content or 'href="{% static' in content + + def test_theme_color_meta(self): + """Test that base.html has theme-color meta tag.""" + base_path = Path("templates/base.html") + with open(base_path) as f: + content = f.read() + assert 'name="theme-color"' in content + + def test_apple_web_app_meta(self): + """Test that base.html has apple-mobile-web-app-capable meta tag.""" + base_path = Path("templates/base.html") + with open(base_path) as f: + content = f.read() + assert 'name="apple-mobile-web-app-capable"' in content + + def test_viewport_meta(self): + """Test that base.html has proper viewport meta tag.""" + base_path = Path("templates/base.html") + with open(base_path) as f: + content = f.read() + assert 'name="viewport"' in content + assert "width=device-width" in content + + +class TestResponsiveCSS: + """Tests for responsive CSS.""" + + def test_css_has_media_queries(self): + """Test that CSS has mobile-first responsive breakpoints.""" + css_path = Path("static/css/base.css") + with open(css_path) as f: + content = f.read() + + # Check for mobile-first breakpoints + assert "@media (max-width:" in content + assert "max-width: 640px" in content or "max-width: 768px" in content + + def test_css_uses_relative_units(self): + """Test that CSS uses relative units (rem/em) instead of fixed px.""" + css_path = Path("static/css/base.css") + with open(css_path) as f: + content = f.read() + + # Check that rem/em are used for spacing/typography + assert "rem" in content or "em" in content + + +class TestLighthouseCI: + """Tests for Lighthouse CI configuration.""" + + def test_lighthouserc_exists(self): + """Test that lighthouserc.json exists.""" + lh_path = Path("lighthouserc.json") + assert lh_path.exists(), "lighthouserc.json should exist" + + def test_lighthouse_config_valid(self): + """Test that lighthouserc.json is valid JSON with PWA config.""" + lh_path = Path("lighthouserc.json") + with open(lh_path) as f: + config = json.load(f) + + assert "ci" in config + assert "collect" in config["ci"] + assert "assert" in config["ci"] + + +class TestGitHubActions: + """Tests for GitHub Actions Lighthouse CI step.""" + + def test_github_actions_workflow_exists(self): + """Test that GitHub Actions workflow exists for Lighthouse CI.""" + workflow_dir = Path(".github/workflows") + assert workflow_dir.exists() + + workflow_files = list(workflow_dir.glob("*.yml")) + list( + workflow_dir.glob("*.yaml") + ) + assert len(workflow_files) > 0, "At least one workflow file should exist" + + def test_lighthouse_step_in_workflow(self): + """Test that Lighthouse CI step exists in a workflow.""" + workflow_dir = Path(".github/workflows") + workflow_files = list(workflow_dir.glob("*.yml")) + list( + workflow_dir.glob("*.yaml") + ) + + found = False + for wf in workflow_files: + with open(wf) as f: + content = f.read() + if "lighthouse" in content.lower() or "lighthouseci" in content: + found = True + break + + assert found, "No Lighthouse CI step found in GitHub Actions workflows" + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_reso_adapter.py b/tests/test_reso_adapter.py new file mode 100644 index 00000000..95285d50 --- /dev/null +++ b/tests/test_reso_adapter.py @@ -0,0 +1,266 @@ +"""Tests for RESO Web API adapter.""" + +import pytest +from decimal import Decimal +from unittest.mock import Mock, patch + +from core.integrations.sources.reso_adapter import ( + RESOAdapter, + normalize_property_type, + normalize_property_data, +) + + +class TestPropertyTypeNormalization: + """Tests for property type normalization.""" + + def test_normalize_sfr(self) -> None: + """Test single family residence normalization.""" + assert normalize_property_type("Single Family") == "SFR" + assert normalize_property_type("Single Family Residence") == "SFR" + assert normalize_property_type("SINGLE FAMILY") == "SFR" + assert normalize_property_type("Single Family Detached") == "SFR" + + def test_normalize_condo(self) -> None: + """Test condo normalization.""" + assert normalize_property_type("Condo") == "CONDO" + assert normalize_property_type("Condominium") == "CONDO" + + def test_normalize_townhouse(self) -> None: + """Test townhouse normalization.""" + assert normalize_property_type("Townhouse") == "TOWNHOUSE" + assert normalize_property_type("Town House") == "TOWNHOUSE" + + def test_normalize_multi(self) -> None: + """Test multi-family normalization.""" + assert normalize_property_type("Duplex") == "DUPLEX" + assert normalize_property_type("Triplex") == "TRIPLEX" + assert normalize_property_type("Fourplex") == "FOURPLEX" + assert normalize_property_type("Multifamily") == "MULTIFAMILY" + + def test_normalize_unknown(self) -> None: + """Test unknown type returns uppercase version.""" + assert normalize_property_type("Weird Type") == "WEIRD TYPE" + assert normalize_property_type(None) is None + assert normalize_property_type("") is None + + +class TestPropertyDataNormalization: + """Tests for property data normalization.""" + + @pytest.mark.django_db + def test_normalize_complete_property(self) -> None: + """Test normalizing complete property data.""" + + raw = { + "ListingId": "12345", + "ListingKey": "LIST-123", + "UnparsedAddress": "123 Main St", + "StreetNumber": "123", + "StreetName": "Main St", + "City": "Austin", + "StateOrProvince": "TX", + "PostalCode": "78701", + "ListPrice": "450000", + "BedroomsTotal": "3", + "BathroomsTotalInteger": "2", + "LivingArea": "2000", + "LotSizeSquareFeet": "7500", + "PropertyType": "Single Family", + "PropertySubType": "Detached", + "YearBuilt": "2010", + "LotSizeAcres": "0.25", + "DaysOnMarket": "15", + "StandardStatus": "Active", + "ListingContractDate": "2024-01-15", + "ExpirationDate": "2024-07-15", + "MlsNumber": "1234567", + "MlsId": "TX-123", + "Latitude": "30.2672", + "Longitude": "-97.7431", + "Media": [ + { + "MediaURL": "https://example.com/photo1.jpg", + "MediaType": "Photo", + "Description": "Front view", + "Order": 1, + }, + { + "MediaURL": "https://example.com/photo2.jpg", + "MediaType": "Photo", + "Order": 2, + }, + ], + "VirtualTourURL": "https://tour.example.com/123", + "PublicRemarks": "Beautiful home in Austin", + "PrivateRemarks": "Agent only remarks", + "ListAgentKey": "AGENT123", + "ListOfficeKey": "OFFICE456", + } + + result = normalize_property_data(raw) + + assert result["source"] == "reso" + assert result["listing_id"] == "12345" + assert result["address"] == "123 Main St" + assert result["city"] == "Austin" + assert result["state"] == "TX" + assert result["zip_code"] == "78701" + assert result["price"] == 450000 + assert result["beds"] == 3 + assert result["baths"] == Decimal("2") + assert result["sq_ft"] == 2000 + assert result["property_type"] == "SFR" + assert result["property_sub_type"] == "Detached" + assert result["year_built"] == 2010 + assert result["lot_size_acres"] == Decimal("0.25") + assert result["days_on_market"] == 15 + assert result["listing_status"] == "Active" + assert len(result["photos"]) == 2 + assert result["photos"][0]["url"] == "https://example.com/photo1.jpg" + assert result["photos"][0]["order"] == 1 + assert result["virtual_tour_url"] == "https://tour.example.com/123" + assert result["remarks"] == "Beautiful home in Austin" + assert result["agent_id"] == "AGENT123" + assert result["office_id"] == "OFFICE456" + + def test_normalize_minimal_property(self) -> None: + """Test normalizing minimal property data.""" + + raw = { + "ListingId": "MIN-001", + "UnparsedAddress": "456 Oak Ave", + "City": "Dallas", + "StateOrProvince": "TX", + "PostalCode": "75201", + "ListPrice": "300000", + } + + result = normalize_property_data(raw) + + assert result["listing_id"] == "MIN-001" + assert result["address"] == "456 Oak Ave" + assert result["city"] == "Dallas" + assert result["state"] == "TX" + assert result["zip_code"] == "75201" + assert result["price"] == 300000 + assert result["property_type"] is None # Not provided + assert result["photos"] == [] + + +class TestRESOAdapter: + """Tests for RESOAdapter class.""" + + @pytest.fixture + def adapter(self) -> RESOAdapter: + """Create a test adapter instance.""" + return RESOAdapter( + base_url="https://api.test.mls.com/odata", + username="test_user", + password="test_pass", + ) + + def test_adapter_creation(self, adapter: RESOAdapter) -> None: + """Test adapter initialization.""" + assert adapter.base_url == "https://api.test.mls.com/odata" + assert adapter.username == "test_user" + assert adapter.password == "test_pass" # noqa: S105 - test fixture credential + + def test_build_filter_simple(self) -> None: + """Test simple filter building.""" + adapter = RESOAdapter(base_url="https://api.test.com/odata") + + filter_expr = adapter.build_filter( + [ + {"field": "City", "operator": "eq", "value": "Austin"}, + {"field": "ListPrice", "operator": "ge", "value": 300000}, + ] + ) + + # The exact format may vary, but should contain both conditions + assert "City eq 'Austin'" in filter_expr + assert "ListPrice ge 300000" in filter_expr + + def test_build_filter_operators(self) -> None: + """Test various filter operators.""" + adapter = RESOAdapter(base_url="https://api.test.com/odata") + + # Test gt + assert "gt 100" in adapter.build_filter( + [{"field": "Price", "operator": "gt", "value": 100}] + ) + + # Test lt + assert "lt 100" in adapter.build_filter( + [{"field": "Price", "operator": "lt", "value": 100}] + ) + + # Test contains + assert "contains" in adapter.build_filter( + [{"field": "City", "operator": "contains", "value": "Austin"}] + ) + + @patch("core.integrations.sources.reso_adapter.requests.Session.request") + def test_fetch_property_success(self, mock_request, adapter: RESOAdapter) -> None: + """Test successful property fetch.""" + mock_response = Mock() + mock_response.json.return_value = { + "ListingId": "12345", + "ListingKey": "LIST-123", + "ListPrice": 450000, + "City": "Austin", + } + mock_response.raise_for_status = Mock() + mock_response.ok = True + mock_response.status_code = 200 + mock_request.return_value = mock_response + + result = adapter.fetch_property("LIST-123") + + assert result["ListingId"] == "12345" + assert result["ListingKey"] == "LIST-123" + assert result["ListPrice"] == 450000 + + @patch("core.integrations.sources.reso_adapter.requests.Session.request") + def test_fetch_property_404(self, mock_request, adapter: RESOAdapter) -> None: + """Test 404 handling.""" + mock_response = Mock() + mock_response.status_code = 404 + mock_response.ok = False + mock_response.text = "Not Found" + mock_response.raise_for_status = Mock(side_effect=Exception("404 Not Found")) + mock_request.return_value = mock_response + + with pytest.raises(Exception) as exc_info: + adapter.fetch_property("NONEXISTENT") + assert ( + "404" in str(exc_info.value) or "not found" in str(exc_info.value).lower() + ) + + @patch("core.integrations.sources.reso_adapter.requests.Session.request") + def test_query_properties(self, mock_request, adapter: RESOAdapter) -> None: + """Test property query with filters.""" + mock_response = Mock() + mock_response.json.return_value = { + "value": [ + {"ListingId": "1", "ListPrice": 300000}, + {"ListingId": "2", "ListPrice": 400000}, + ], + "@odata.count": 2, + } + mock_response.raise_for_status = Mock() + mock_response.ok = True + mock_response.status_code = 200 + mock_request.return_value = mock_response + + result = adapter.query_properties( + filter_expr="City eq 'Austin'", + top=2, + ) + + assert len(result["value"]) == 2 + assert result["@odata.count"] == 2 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/test_saved_searches_notifications.py b/tests/test_saved_searches_notifications.py new file mode 100644 index 00000000..4e762280 --- /dev/null +++ b/tests/test_saved_searches_notifications.py @@ -0,0 +1,155 @@ +"""Tests for saved searches with email/in-app notifications.""" + +import pytest +from decimal import Decimal + +from core.models.growth import SavedSearch +from core.services.saved_search_notifications import ( + check_listings_against_saved_searches, + get_saved_search_matches, + create_notification_for_match, +) + + +class TestSavedSearchMatches: + """Tests for checking listings against saved searches.""" + + def test_get_saved_search_matches(self) -> None: + """Test that saved search returns matching listings.""" + + saved_search = SavedSearch( + name="Test Search", + state="TX", + zip_code="78701", + min_price=Decimal("300000"), + max_price=Decimal("500000"), + ) + + # Mock listings data (would normally come from external API) + listings = [ + {"price": Decimal("350000"), "state": "TX", "zip_code": "78701"}, + {"price": Decimal("450000"), "state": "TX", "zip_code": "78701"}, + { + "price": Decimal("250000"), + "state": "TX", + "zip_code": "78701", + }, # Below min_price + { + "price": Decimal("550000"), + "state": "TX", + "zip_code": "78701", + }, # Above max_price + ] + + matches = get_saved_search_matches(saved_search, listings) + + # Should only match listings within price range + assert len(matches) == 2 + + def test_get_saved_search_matches_empty(self) -> None: + """Test that no matches returns empty list.""" + + saved_search = SavedSearch( + name="Test Search", + state="TX", + zip_code="78701", + min_price=Decimal("300000"), + max_price=Decimal("500000"), + ) + + listings = [ + {"price": Decimal("250000"), "state": "TX", "zip_code": "78701"}, + {"price": Decimal("550000"), "state": "TX", "zip_code": "78701"}, + ] + + matches = get_saved_search_matches(saved_search, listings) + assert len(matches) == 0 + + +@pytest.mark.django_db +class TestNotificationCreation: + """Tests for creating notifications for matches.""" + + def test_create_notification_for_match(self) -> None: + """Test that notification is created for a match.""" + from core.models import User + + user = User.objects.create_user( + username="testuser", email="test@example.com", password="testpass123" + ) + + saved_search = SavedSearch.objects.create( + user=user, + name="Downtown Search", + state="TX", + zip_code="78701", + min_price=Decimal("300000"), + max_price=Decimal("500000"), + ) + + listing = { + "price": Decimal("350000"), + "state": "TX", + "zip_code": "78701", + "address": "123 Main St", + } + + notification = create_notification_for_match(saved_search, listing) + + assert notification is not None + assert notification.user == user + assert "Downtown Search" in notification.title + + +@pytest.mark.django_db +class TestCheckListings: + """Tests for checking listings against saved searches.""" + + def test_check_listings_creates_notifications(self) -> None: + """Test that checking listings creates notifications for matches.""" + from core.models import User + from unittest.mock import patch + + user = User.objects.create_user( + username="testuser2", email="test2@example.com", password="testpass123" + ) + + SavedSearch.objects.create( + user=user, + name="Test Search", + state="TX", + zip_code="78701", + min_price=Decimal("300000"), + max_price=Decimal("500000"), + ) + + # Mock the listings data + mock_listings = [ + {"price": Decimal("350000"), "state": "TX", "zip_code": "78701"}, + {"price": Decimal("450000"), "state": "TX", "zip_code": "78701"}, + ] + + with patch( + "core.services.saved_search_notifications.fetch_new_listings", + return_value=mock_listings, + ): + result = check_listings_against_saved_searches() + + assert result["searches_checked"] >= 1 + assert result["matches_found"] >= 2 + + +class TestSavedSearchAdmin: + """Tests for saved search admin configuration.""" + + def test_saved_search_admin_fields(self) -> None: + """Test that SavedSearchAdmin has correct fields.""" + from core.admin import SavedSearchAdmin + + assert "user" in SavedSearchAdmin.list_display + assert "name" in SavedSearchAdmin.list_display + assert "state" in SavedSearchAdmin.list_display + assert "zip_code" in SavedSearchAdmin.list_display + assert "min_price" in SavedSearchAdmin.list_display + assert "max_price" in SavedSearchAdmin.list_display + assert "created_at" in SavedSearchAdmin.list_display diff --git a/tests/test_tax_strategy.py b/tests/test_tax_strategy.py new file mode 100644 index 00000000..413e8bca --- /dev/null +++ b/tests/test_tax_strategy.py @@ -0,0 +1,196 @@ +"""Tests for tax strategy module (QBI, PAL, 1031 Exchange).""" + +from decimal import Decimal + +from investor_app.finance.tax_strategy import ( + calculate_qbi_deduction, + calculate_pal_allowance, + calculate_1031_exchange, + calculate_1031_deferral_ratio, + calculate_pal_phase_out, + calculate_total_tax_benefit, + PAL_FULL_ALLOWANCE, +) + + +class TestQBI: + """Tests for Qualified Business Income (QBI) deduction.""" + + def test_qbi_deduction_basic(self) -> None: + """Test basic QBI deduction calculation.""" + result = calculate_qbi_deduction( + qualified_business_income=Decimal("50000"), + w2_wages=Decimal("0"), + qbi_adjusted_basis=Decimal("0"), + ) + # 20% of $50,000 = $10,000 + assert result == Decimal("10000.00") + + def test_qbi_deduction_with_w2_limit(self) -> None: + """Test QBI deduction with W-2 wage limitation.""" + # For high-income taxpayers, QBI deduction is limited to 50% of W-2 wages + result = calculate_qbi_deduction( + qualified_business_income=Decimal("200000"), + w2_wages=Decimal("50000"), + qbi_adjusted_basis=Decimal("0"), + taxable_income=Decimal("300000"), # Above phase-out + ) + # 50% of W-2 wages = $25,000 (which is less than 20% of income = $40,000) + assert result == Decimal("25000.00") + + def test_qbi_deduction_with_basis_limit(self) -> None: + """Test QBI deduction with qualified property limitation.""" + # 2.5% of QBI adjusted basis + result = calculate_qbi_deduction( + qualified_business_income=Decimal("100000"), + w2_wages=Decimal("0"), + qbi_adjusted_basis=Decimal("400000"), + taxable_income=Decimal("300000"), + ) + # 2.5% of $400,000 = $10,000 + assert result == Decimal("10000.00") + + def test_qbi_deduction_below_phaseout(self) -> None: + """Test QBI deduction for income below phase-out threshold.""" + result = calculate_qbi_deduction( + qualified_business_income=Decimal("100000"), + w2_wages=Decimal("30000"), + qbi_adjusted_basis=Decimal("200000"), + taxable_income=Decimal("150000"), # Below $164,900 for single + ) + # No phase-out, full 20% deduction + assert result == Decimal("20000.00") + + def test_qbi_deduction_zero_income(self) -> None: + """Test QBI deduction with zero income.""" + result = calculate_qbi_deduction( + qualified_business_income=Decimal("0"), + w2_wages=Decimal("0"), + qbi_adjusted_basis=Decimal("0"), + ) + assert result == Decimal("0") + + +class TestPAL: + """Tests for Passive Activity Loss (PAL) rules.""" + + def test_pal_full_allowance_below_threshold(self) -> None: + """Test PAL allowance for income below $100k threshold.""" + result = calculate_pal_allowance( + active_participation=Decimal("1"), + modified_agi=Decimal("80000"), + rental_losses=Decimal("25000"), + ) + # Full $25,000 allowance for active participation + assert result == PAL_FULL_ALLOWANCE + + def test_pal_zero_for_inactive_participation(self) -> None: + """Test PAL allowance is zero for non-active participation.""" + result = calculate_pal_allowance( + active_participation=Decimal("0"), + modified_agi=Decimal("80000"), + rental_losses=Decimal("25000"), + ) + assert result == Decimal("0") + + def test_pal_phase_out(self) -> None: + """Test PAL phase-out between $100k-$150k AGI.""" + # $125k AGI = midpoint between $100k and $150k = 50% reduction + phase_out = calculate_pal_phase_out(Decimal("125000")) + assert phase_out == Decimal("0.50") + + def test_pal_phase_out_below_threshold(self) -> None: + """Test PAL phase-out below $100k AGI.""" + phase_out = calculate_pal_phase_out(Decimal("80000")) + assert phase_out == Decimal("1") + + def test_pal_phase_out_above_threshold(self) -> None: + """Test PAL phase-out above $150k AGI.""" + phase_out = calculate_pal_phase_out(Decimal("200000")) + assert phase_out == Decimal("0") + + def test_pal_max_losses(self) -> None: + """Test PAL deductibility capped at losses.""" + result = calculate_pal_allowance( + active_participation=Decimal("1"), + modified_agi=Decimal("50000"), + rental_losses=Decimal("30000"), # Greater than $25k cap + ) + # Capped at $25,000 maximum + assert result == PAL_FULL_ALLOWANCE + + +class Test1031Exchange: + """Tests for 1031 Exchange calculations.""" + + def test_basic_deferral(self) -> None: + """Test basic 1031 exchange deferral.""" + result = calculate_1031_exchange( + sale_price=Decimal("500000"), + original_cost_basis=Decimal("300000"), + accumulated_depreciation=Decimal("45000"), + replacement_price=Decimal("600000"), + selling_costs=Decimal("30000"), + ) + # All gains deferred if replacement price >= sale price + assert result["deferred_gain"] == Decimal("200000") + + def test_partial_deferral(self) -> None: + """Test 1031 exchange with partial deferral.""" + result = calculate_1031_exchange( + sale_price=Decimal("500000"), + original_cost_basis=Decimal("300000"), + accumulated_depreciation=Decimal("45000"), + replacement_price=Decimal("400000"), + selling_costs=Decimal("30000"), + ) + # Partial deferral: replacement price < sale price + assert result["deferred_gain"] < Decimal("200000") + assert result["boot_received"] > Decimal("0") + + def test_deferral_ratio(self) -> None: + """Test deferral ratio calculation.""" + ratio = calculate_1031_deferral_ratio( + sale_price=Decimal("500000"), + replacement_price=Decimal("500000"), + selling_costs=Decimal("30000"), + ) + # 100% deferral if replacement price covers net sale price + assert ratio == Decimal("1") + + def test_partial_deferral_ratio(self) -> None: + """Test partial deferral ratio.""" + ratio = calculate_1031_deferral_ratio( + sale_price=Decimal("500000"), + replacement_price=Decimal("450000"), + selling_costs=Decimal("30000"), + ) + # 450000 / (500000 - 30000) = 0.957... + expected = Decimal("450000") / (Decimal("500000") - Decimal("30000")) + assert abs(ratio - expected) < Decimal("0.01") + + def test_total_tax_benefit(self) -> None: + """Test total tax benefit calculation.""" + benefit = calculate_total_tax_benefit( + qbi_income=Decimal("50000"), + rental_losses=Decimal("20000"), + modified_agi=Decimal("80000"), + marginal_tax_rate=Decimal("0.24"), + ) + # QBI deduction = 20% of $50,000 = $10,000 + # PAL = $20,000 (full allowance) + # Total tax benefit = ($10,000 + $20,000) * 0.24 = $7,200 + assert benefit == Decimal("7200.00") + + def test_total_tax_benefit_with_phase_out(self) -> None: + """Test total tax benefit with PAL phase-out.""" + benefit = calculate_total_tax_benefit( + qbi_income=Decimal("100000"), + rental_losses=Decimal("25000"), + modified_agi=Decimal("125000"), # Midpoint = 50% reduction + marginal_tax_rate=Decimal("0.24"), + ) + # QBI deduction = 20% of $100,000 = $20,000 + # PAL = $25,000 * 0.50 = $12,500 (50% reduction) + # Total tax benefit = ($20,000 + $12,500) * 0.24 = $7,800 + assert benefit == Decimal("7800.00")