Spaces:
Sleeping
Sleeping
Z User
feat: ABM Engine v1 — OpenSearch store + Apify enrichment + ICP scoring + k-NN memory + optional Neo4j mirror
fb0ec21 | """ICP scoring — deterministic, explainable account prioritization. | |
| Score 0-100 from additive signals with the breakdown preserved, so sales can | |
| see WHY an account scored what it scored. Labels: A ≥ 75, B ≥ 50, C ≥ 30. | |
| """ | |
| from typing import Any, Dict | |
| from ..config import settings | |
| from ..stores import embeddings | |
| def score_account(account: Dict[str, Any], contact_count: int = 0) -> Dict[str, Any]: | |
| breakdown: Dict[str, float] = {} | |
| industry = str(account.get("industry") or "").lower() | |
| hit = next((x for x in settings.icp_industries_list if x and x in industry), None) | |
| breakdown["industry_fit"] = 20.0 if hit else 0.0 | |
| country = str(account.get("hq_country") or "").lower() | |
| breakdown["geo_fit"] = 15.0 if any(x in country for x in settings.icp_countries_list if x) else 0.0 | |
| headcount = account.get("employee_count") | |
| if isinstance(headcount, str) and headcount.isdigit(): | |
| headcount = int(headcount) | |
| if isinstance(headcount, (int, float)) and headcount: | |
| lo, hi = settings.icp_min_headcount, settings.icp_max_headcount | |
| breakdown["size_band"] = 15.0 if lo <= headcount <= hi else 5.0 | |
| else: | |
| breakdown["size_band"] = 0.0 | |
| breakdown["digital_presence"] = 15.0 if (account.get("domain") or account.get("website")) else 0.0 | |
| rating = account.get("rating") | |
| if isinstance(rating, (int, float)) and rating: | |
| # Google rating (1-5) → up to 10 points; presence alone is a weak signal | |
| breakdown["market_signal"] = min(max((rating - 3.0) / 2.0, 0.0), 1.0) * 10.0 | |
| elif isinstance(rating, str) and rating.replace(".", "").isdigit(): | |
| breakdown["market_signal"] = min(max((float(rating) - 3.0) / 2.0, 0.0), 1.0) * 10.0 | |
| else: | |
| breakdown["market_signal"] = 0.0 | |
| breakdown["contact_coverage"] = min(contact_count, 5) * 2.0 # up to 10 | |
| score = round(min(sum(breakdown.values()), 100.0), 1) | |
| label = "A" if score >= 75 else "B" if score >= 50 else "C" if score >= 30 else "D" | |
| return { | |
| "icp_score": score, | |
| "icp_label": label, | |
| "breakdown": {k: round(v, 1) for k, v in breakdown.items()}, | |
| "scoring_version": "icp-v1", | |
| } | |
| def account_text(account: Dict[str, Any]) -> str: | |
| """The text that represents an account in the vector memory.""" | |
| parts = [ | |
| account.get("name") or "", | |
| account.get("industry") or "", | |
| account.get("description") or "", | |
| f"domain {account.get('domain')}" if account.get("domain") else "", | |
| f"city {account.get('city')}" if account.get("city") else "", | |
| f"country {account.get('hq_country')}" if account.get("hq_country") else "", | |
| ] | |
| return ". ".join(p for p in parts if p) | |
| def embed_account(account: Dict[str, Any]): | |
| return embeddings.embed(account_text(account)) | |