Wholesale and distribution

Data mapping

Entity

account

Group

sales region

Profile

b2b

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": "Cuenta", "time": "Fecha Factura", "type": "Tipo Doc", "value": "Net Value", "group": "Sales Org", "attrs": ["Payment Terms", "Sales Rep", "Due Date", "Paid Date"], "date_format": "%d.%m.%Y"})
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("b2b")
def credit_notes_180(entity, before, cut):   # receives only events recorded before the cutoff
    return {"crn_180": sum(1 for e in before if e.type == "CRN"
                          and e.at >= cut - timedelta(days=180))}

task = predict_event(ds, target="INV", window_days=90, pack="b2b")
print(task.status, task.report())     # "trained" only if the model outperformed the baseline on held-out entities
if task.status == "trained":
    print(task.score("ACC4399"))  # probability with contributing factors

size = predict_value(ds, target="INV", window_days=90, pack="b2b")
print(size.status, size.report())

flags = rules.run(ds, [
    rules.overdue("Due Date", "Paid Date", tolerance_days=15),   # invoices paid more than 15 days after the due date
])
print(flags["summary"])

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