Built a lead scoring system in Flux to replace a brittle set of spreadsheets and manual updates. Goal was to automate scoring based on website visits, content downloads, and demo requests, pushing results to our CRM.
The core logic sits in a few key `Dataframe` operations. Here's the main scoring transformation:
```python
scored_leads = (
raw_lead_events
.groupby("lead_id")
.agg(
visit_score=("page_views", "sum"),
content_score=("whitepaper_downloads", "count") * 5,
demo_score=("demo_requested", "max") * 20
)
.map(lambda x: x.fillna(0))
.with_column(
"total_score",
pl.col("visit_score") + pl.col("content_score") + pl.col("demo_score")
)
.filter(pl.col("total_score") > 25) # Threshold for MQL
)
```
A separate flow handles the CRM sync via webhook. Main benefit is the transparency; the entire scoring logic is now in version control and updates with a PR. It's been running for three weeks with zero intervention.
Nice to see scoring logic defined in code rather than hidden in formulas. Have you considered the cardinality of `lead_id`? If it's high, that groupby-aggregate could get heavy. Might be worth pushing the threshold filter earlier, maybe even pre-aggregate, to reduce the working set.
Also, watch out for that `max()` on a boolean for demo_score. It works, but semantically it's a bit opaque. Could you filter for `demo_requested == true` and count distinct lead_ids instead? Might map cleaner to the business rule.
How often does this flow run? If it's frequent, you'll want to think about incremental aggregation, only processing new events. The current approach recalculates all history every time.
sub-100ms or bust