Retail and consumer goods

Data mapping

Entity

customer

Group

store or channel

Profile

retail

Reference implementation

from datetime import datetime, timedelta
from vertiq import plugins, rules
from vertiq.mapping import Mapping, apply, repair
from vertiq.tasks import predict_event, predict_value

# rows: the source export as a list of dicts, e.g. list(csv.DictReader(open("export.csv")))
mapping = Mapping.parse({"entity": "Customer ID", "time": "Order Date", "type": "Order Type", "value": "Order Value", "group": "Store", "attrs": ["Category", "Coupon"], "date_format": "%Y-%m-%d"})
m, notes = repair(mapping, rows)      # validates column names, value types and date format against the rows
ds, problems = apply(m, rows)         # rows that fail to load are reported, not imputed

@plugins.feature("retail")
def return_rate(entity, before, cut):   # receives only events recorded before the cutoff
    orders = [e for e in before if e.type == "order"]
    returns = [e for e in before if e.type == "return"]
    return {"return_share": len(returns) / max(1, len(orders))}

task = predict_event(ds, target="order", window_days=60, pack="retail")
print(task.status, task.report())     # "trained" only if the model outperformed the baseline on held-out entities
if task.status == "trained":
    print(task.score("C-88213"))  # probability with contributing factors

size = predict_value(ds, target="order", window_days=60, pack="retail")
print(size.status, size.report())

flags = rules.run(ds, [
    rules.repeat(1, type="return"),   # two returns by the same customer on one day
])
print(flags["summary"])

This code is executed by the test suite (tests/test_guides.py) against generated records with these column names.