Aissa
5d60d33db1
Volledige Foodlinkk Command Center uitbreiding met social automatisering, reclamefolder filters, Proxmox monitoring en documentatie.
109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
"""Halal market opportunity scoring for retail locations."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.db import execute, fetch_all, fetch_one, json_param
|
|
|
|
|
|
def compute_all_scores(limit: int = 5000) -> dict[str, Any]:
|
|
rows = fetch_all(
|
|
"""
|
|
SELECT s.id, s.chain, s.city, s.postcode, s.halal_certified, s.has_halal_section,
|
|
s.partnership_status, a.population, a.avg_income,
|
|
(a.religious_composition->>'muslim_proxy_pct')::float AS muslim_pct,
|
|
(a.ethnic_composition->>'niet_westers_pct')::float AS niet_westers_pct
|
|
FROM supermarkets s
|
|
LEFT JOIN area_analysis a ON a.postcode = s.postcode
|
|
WHERE s.postcode <> '0000AA'
|
|
LIMIT %s
|
|
""",
|
|
(limit,),
|
|
)
|
|
computed = 0
|
|
for r in rows:
|
|
muslim = float(r.get("muslim_pct") or 0)
|
|
niet_w = float(r.get("niet_westers_pct") or 0)
|
|
pop = int(r.get("population") or 0)
|
|
income = float(r.get("avg_income") or 0)
|
|
|
|
halal_gap = 0.0
|
|
if not r.get("halal_certified") and not r.get("has_halal_section"):
|
|
halal_gap = min(100, muslim * 1.5 + niet_w * 0.5)
|
|
elif r.get("has_halal_section") and not r.get("halal_certified"):
|
|
halal_gap = min(80, muslim * 0.8)
|
|
|
|
market_potential = 0.0
|
|
if pop > 0:
|
|
market_potential += min(40, pop / 25000)
|
|
if income > 0:
|
|
market_potential += min(30, income / 1500)
|
|
market_potential += min(30, muslim * 0.4)
|
|
|
|
partnership_bonus = 15 if r.get("partnership_status") == "active" else 0
|
|
halal_opp = round(min(100, halal_gap + market_potential * 0.3), 1)
|
|
market_score = round(min(100, market_potential + partnership_bonus), 1)
|
|
|
|
factors = {
|
|
"muslim_proxy_pct": muslim,
|
|
"niet_westers_pct": niet_w,
|
|
"population": pop,
|
|
"avg_income": income,
|
|
"halal_gap": round(halal_gap, 1),
|
|
"has_halal_section": bool(r.get("has_halal_section")),
|
|
"halal_certified": bool(r.get("halal_certified")),
|
|
"partnership_status": r.get("partnership_status"),
|
|
}
|
|
execute(
|
|
"""
|
|
INSERT INTO retail_opportunity_scores (supermarket_id, halal_opportunity_score,
|
|
market_potential_score, factors, computed_at)
|
|
VALUES (%s, %s, %s, %s, NOW())
|
|
ON CONFLICT (supermarket_id) DO UPDATE SET
|
|
halal_opportunity_score = EXCLUDED.halal_opportunity_score,
|
|
market_potential_score = EXCLUDED.market_potential_score,
|
|
factors = EXCLUDED.factors,
|
|
computed_at = NOW()
|
|
""",
|
|
(r["id"], halal_opp, market_score, json_param(factors)),
|
|
)
|
|
execute(
|
|
"UPDATE supermarkets SET halal_opportunity_score = %s WHERE id = %s",
|
|
(halal_opp, r["id"]),
|
|
)
|
|
computed += 1
|
|
return {"computed": computed}
|
|
|
|
|
|
def top_opportunities(
|
|
limit: int = 50,
|
|
min_score: float = 30,
|
|
chain: str | None = None,
|
|
province: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
clauses = ["ros.halal_opportunity_score >= %s"]
|
|
params: list[Any] = [min_score]
|
|
if chain:
|
|
clauses.append("s.chain ILIKE %s")
|
|
params.append(f"%{chain}%")
|
|
if province:
|
|
clauses.append("s.province ILIKE %s")
|
|
params.append(f"%{province}%")
|
|
where = " AND ".join(clauses)
|
|
return fetch_all(
|
|
f"""
|
|
SELECT s.id, s.name, s.chain, s.city, s.province, s.postcode,
|
|
s.partnership_status, s.halal_certified, s.has_halal_section,
|
|
ros.halal_opportunity_score, ros.market_potential_score, ros.factors,
|
|
a.population, a.avg_income,
|
|
(a.religious_composition->>'muslim_proxy_pct')::float AS muslim_proxy_pct
|
|
FROM retail_opportunity_scores ros
|
|
JOIN supermarkets s ON s.id = ros.supermarket_id
|
|
LEFT JOIN area_analysis a ON a.postcode = s.postcode
|
|
WHERE {where}
|
|
ORDER BY ros.halal_opportunity_score DESC
|
|
LIMIT %s
|
|
""",
|
|
tuple(params + [limit]),
|
|
)
|