270 lines
9.4 KiB
Python
270 lines
9.4 KiB
Python
"""
|
|
Competitor Analyzer
|
|
|
|
Specialized analysis for competitor bakeries with competitive pressure modeling.
|
|
Treats competitor proximity differently than other POIs, considering market dynamics.
|
|
"""
|
|
|
|
from typing import Dict, List, Any, Tuple
|
|
import structlog
|
|
from math import radians, sin, cos, sqrt, atan2
|
|
|
|
from app.core.poi_config import COMPETITOR_ZONES
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class CompetitorAnalyzer:
|
|
"""
|
|
Competitive landscape analyzer for bakery locations.
|
|
|
|
Models competitive pressure considering:
|
|
- Direct competition (<100m): Strong negative impact
|
|
- Nearby competition (100-500m): Moderate negative impact
|
|
- Market saturation (500-1000m): Can be positive (bakery district)
|
|
or negative (competitive market)
|
|
"""
|
|
|
|
def analyze_competitive_landscape(
|
|
self,
|
|
competitor_pois: List[Dict[str, Any]],
|
|
bakery_location: Tuple[float, float],
|
|
tenant_id: str = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Analyze competitive pressure from nearby bakeries.
|
|
|
|
Args:
|
|
competitor_pois: List of detected competitor POIs
|
|
bakery_location: Tuple of (latitude, longitude)
|
|
tenant_id: Optional tenant ID for logging
|
|
|
|
Returns:
|
|
Competitive analysis with pressure scores and market classification
|
|
"""
|
|
if not competitor_pois:
|
|
logger.info(
|
|
"No competitors detected - underserved market",
|
|
tenant_id=tenant_id
|
|
)
|
|
return {
|
|
"competitive_pressure_score": 0.0,
|
|
"direct_competitors_count": 0,
|
|
"nearby_competitors_count": 0,
|
|
"market_competitors_count": 0,
|
|
"competitive_zone": "low_competition",
|
|
"market_type": "underserved",
|
|
"competitive_advantage": "first_mover",
|
|
"ml_feature_competitive_pressure": 0.0,
|
|
"ml_feature_has_direct_competitor": 0,
|
|
"ml_feature_competitor_density_500m": 0,
|
|
"competitor_details": []
|
|
}
|
|
|
|
# Categorize competitors by distance
|
|
direct_competitors = [] # <100m
|
|
nearby_competitors = [] # 100-500m
|
|
market_competitors = [] # 500-1000m
|
|
competitor_details = []
|
|
|
|
for poi in competitor_pois:
|
|
distance_m = self._calculate_distance(
|
|
bakery_location, (poi["lat"], poi["lon"])
|
|
) * 1000
|
|
|
|
competitor_info = {
|
|
"name": poi.get("name", "Unnamed"),
|
|
"osm_id": poi.get("osm_id"),
|
|
"distance_m": round(distance_m, 1),
|
|
"lat": poi["lat"],
|
|
"lon": poi["lon"]
|
|
}
|
|
|
|
if distance_m < COMPETITOR_ZONES["direct"]["max_distance_m"]:
|
|
direct_competitors.append(poi)
|
|
competitor_info["zone"] = "direct"
|
|
elif distance_m < COMPETITOR_ZONES["nearby"]["max_distance_m"]:
|
|
nearby_competitors.append(poi)
|
|
competitor_info["zone"] = "nearby"
|
|
elif distance_m < COMPETITOR_ZONES["market"]["max_distance_m"]:
|
|
market_competitors.append(poi)
|
|
competitor_info["zone"] = "market"
|
|
|
|
competitor_details.append(competitor_info)
|
|
|
|
# Calculate competitive pressure score
|
|
direct_pressure = (
|
|
len(direct_competitors) *
|
|
COMPETITOR_ZONES["direct"]["pressure_multiplier"]
|
|
)
|
|
nearby_pressure = (
|
|
len(nearby_competitors) *
|
|
COMPETITOR_ZONES["nearby"]["pressure_multiplier"]
|
|
)
|
|
|
|
# Market saturation analysis
|
|
min_for_district = COMPETITOR_ZONES["market"]["min_count_for_district"]
|
|
if len(market_competitors) >= min_for_district:
|
|
# Many bakeries = destination area (bakery district)
|
|
market_pressure = COMPETITOR_ZONES["market"]["district_multiplier"]
|
|
market_type = "bakery_district"
|
|
elif len(market_competitors) > 2:
|
|
market_pressure = COMPETITOR_ZONES["market"]["normal_multiplier"]
|
|
market_type = "competitive_market"
|
|
else:
|
|
market_pressure = 0.0
|
|
market_type = "normal_market"
|
|
|
|
competitive_pressure_score = (
|
|
direct_pressure + nearby_pressure + market_pressure
|
|
)
|
|
|
|
# Determine competitive zone classification
|
|
if len(direct_competitors) > 0:
|
|
competitive_zone = "high_competition"
|
|
competitive_advantage = "differentiation_required"
|
|
elif len(nearby_competitors) > 2:
|
|
competitive_zone = "moderate_competition"
|
|
competitive_advantage = "quality_focused"
|
|
else:
|
|
competitive_zone = "low_competition"
|
|
competitive_advantage = "local_leader"
|
|
|
|
# Sort competitors by distance
|
|
competitor_details.sort(key=lambda x: x["distance_m"])
|
|
|
|
logger.info(
|
|
"Competitive analysis complete",
|
|
tenant_id=tenant_id,
|
|
competitive_zone=competitive_zone,
|
|
market_type=market_type,
|
|
total_competitors=len(competitor_pois),
|
|
direct=len(direct_competitors),
|
|
nearby=len(nearby_competitors),
|
|
market=len(market_competitors),
|
|
pressure_score=competitive_pressure_score
|
|
)
|
|
|
|
return {
|
|
# Summary scores
|
|
"competitive_pressure_score": round(competitive_pressure_score, 2),
|
|
|
|
# Competitor counts by zone
|
|
"direct_competitors_count": len(direct_competitors),
|
|
"nearby_competitors_count": len(nearby_competitors),
|
|
"market_competitors_count": len(market_competitors),
|
|
"total_competitors_count": len(competitor_pois),
|
|
|
|
# Market classification
|
|
"competitive_zone": competitive_zone,
|
|
"market_type": market_type,
|
|
"competitive_advantage": competitive_advantage,
|
|
|
|
# ML features (for model integration)
|
|
"ml_feature_competitive_pressure": round(competitive_pressure_score, 2),
|
|
"ml_feature_has_direct_competitor": 1 if len(direct_competitors) > 0 else 0,
|
|
"ml_feature_competitor_density_500m": (
|
|
len(direct_competitors) + len(nearby_competitors)
|
|
),
|
|
|
|
# Detailed competitor information
|
|
"competitor_details": competitor_details,
|
|
|
|
# Nearest competitor
|
|
"nearest_competitor": competitor_details[0] if competitor_details else None
|
|
}
|
|
|
|
def _calculate_distance(
|
|
self,
|
|
coord1: Tuple[float, float],
|
|
coord2: Tuple[float, float]
|
|
) -> float:
|
|
"""
|
|
Calculate Haversine distance in kilometers.
|
|
|
|
Args:
|
|
coord1: Tuple of (latitude, longitude)
|
|
coord2: Tuple of (latitude, longitude)
|
|
|
|
Returns:
|
|
Distance in kilometers
|
|
"""
|
|
lat1, lon1 = coord1
|
|
lat2, lon2 = coord2
|
|
|
|
R = 6371 # Earth radius in km
|
|
|
|
dlat = radians(lat2 - lat1)
|
|
dlon = radians(lon2 - lon1)
|
|
|
|
a = (sin(dlat/2)**2 +
|
|
cos(radians(lat1)) * cos(radians(lat2)) * sin(dlon/2)**2)
|
|
c = 2 * atan2(sqrt(a), sqrt(1-a))
|
|
|
|
return R * c
|
|
|
|
def get_competitive_insights(
|
|
self,
|
|
analysis_result: Dict[str, Any]
|
|
) -> List[str]:
|
|
"""
|
|
Generate human-readable competitive insights.
|
|
|
|
Args:
|
|
analysis_result: Result from analyze_competitive_landscape
|
|
|
|
Returns:
|
|
List of insight strings for business intelligence
|
|
"""
|
|
insights = []
|
|
|
|
zone = analysis_result["competitive_zone"]
|
|
market = analysis_result["market_type"]
|
|
pressure = analysis_result["competitive_pressure_score"]
|
|
direct = analysis_result["direct_competitors_count"]
|
|
nearby = analysis_result["nearby_competitors_count"]
|
|
|
|
# Zone-specific insights
|
|
if zone == "high_competition":
|
|
insights.append(
|
|
f"⚠️ High competition: {direct} direct competitor(s) within 100m. "
|
|
"Focus on differentiation and quality."
|
|
)
|
|
elif zone == "moderate_competition":
|
|
insights.append(
|
|
f"Moderate competition: {nearby} nearby competitor(s) within 500m. "
|
|
"Good opportunity for market share."
|
|
)
|
|
else:
|
|
insights.append(
|
|
"✅ Low competition: Local market leader opportunity."
|
|
)
|
|
|
|
# Market type insights
|
|
if market == "bakery_district":
|
|
insights.append(
|
|
"📍 Bakery district: High foot traffic area with multiple bakeries. "
|
|
"Customers actively seek bakery products here."
|
|
)
|
|
elif market == "competitive_market":
|
|
insights.append(
|
|
"Market has multiple bakeries. Quality and customer service critical."
|
|
)
|
|
elif market == "underserved":
|
|
insights.append(
|
|
"🎯 Underserved market: Potential for strong customer base growth."
|
|
)
|
|
|
|
# Pressure score insight
|
|
if pressure < -1.5:
|
|
insights.append(
|
|
"Strong competitive pressure expected to impact demand. "
|
|
"Marketing and differentiation essential."
|
|
)
|
|
elif pressure > 0:
|
|
insights.append(
|
|
"Positive market dynamics: Location benefits from bakery destination traffic."
|
|
)
|
|
|
|
return insights
|