Build a Governed Metric for Revenue/Active Customer/Retention and Trace It to Facts, Dimensions, and Sources

gross_revenue_usd.v1  erp.order_lines.amount_cents  -- measure source  erp.orders.status             -- paid filter source  -> fact_sales.amount_cents  -> semantic metric gross_revenue_usd.v1  -> dashboard / API / notebook consumeractive_customers.v1  erp.orders.customer_id  -> fact_sales.customer_id  -> COUNT(DISTINCT customer_id) within governed UTC date windowperiod_retention.v1  fact_sales.customer_id + order_date  -> prior active set + current active set  -> intersection / prior active

from __future__ import annotationsimport hashlib, json, sqlite3, sysfrom pathlib import PathOUT = Path('/mnt/data/atlasmart_ch20_lab')OUT.mkdir(parents=True, exist_ok=True)DB = OUT / 'atlasmart_ch20.sqlite'if DB.exists(): DB.unlink()# Chapter 19 post-refresh truth, source sequence 207.SALES = [    ('O1001',1,'2026-09-18','P100','C001','web',2,10000,6000,'USD','paid',121),    ('O1001',2,'2026-09-18','P200','C001','web',1,2500,1500,'USD','paid',122),    ('O1002',1,'2026-09-18','P300','C002','mobile',1,19000,12000,'USD','paid',206),    ('O1003',1,'2026-09-19','P400','C001','web',1,10000,6500,'USD','paid',123),    ('O1005',1,'2026-09-20','P100','C004','mobile',1,5000,3000,'USD','paid',125),    ('O1005',2,'2026-09-20','P400','C004','mobile',1,10000,6500,'USD','paid',126),    ('O1007',1,'2026-09-21','P300','C003','sales',1,7500,4500,'USD','paid',201),    ('O1008',1,'2026-09-21','P200','C002','mobile',2,5000,2500,'USD','paid',202),    ('O1009',1,'2026-09-22','P100','C005','web',1,5000,2500,'USD','paid',203),    ('O1010',1,'2026-09-22','P100','C003','sales',1,8000,4500,'USD','paid',207),]CUSTOMERS = [    ('C001','D-CUST-001','Ada Retail','G-NORTH','active'),    ('C002','D-CUST-002','Ben Home','G-EAST','active'),    ('C003','D-CUST-003','Cyra Labs','G-EAST','active'),    ('C004','D-CUST-004','Dara Studio','G-SOUTH','inactive'),    ('C005','D-CUST-005','__INFERRED__','__UNKNOWN__','active'),]DATES = [    ('2026-09-18',2026,9,18,'Friday'),('2026-09-19',2026,9,19,'Saturday'),    ('2026-09-20',2026,9,20,'Sunday'),('2026-09-21',2026,9,21,'Monday'),    ('2026-09-22',2026,9,22,'Tuesday')]PRODUCTS = [    ('P100','Keyboard','Accessories'),('P200','Mouse','Accessories'),    ('P300','Monitor','Displays'),('P400','Dock','Accessories')]EXPECTED_CONTROLS = {'lines':10,'orders':8,'units':12,'revenue_usd':820,'cost_usd':495,'profit_usd':325}METRICS = {  'gross_revenue_usd.v1': {    'type':'metric','measure':'sum(fact_sales.amount_cents)','aggregation':'sum','format':'USD',    'filters':['fact_sales.status = paid','fact_sales.currency_code = USD'],    'time_dimension':'order_date','timezone':'UTC','owner':'finance-analytics',    'grain':'query-context over one current paid order line',    'description':'Gross paid sales amount; returns are not subtracted.',  },  'active_customers.v1': {    'type':'metric','measure':'count_distinct(fact_sales.customer_id)','aggregation':'count_distinct',    'filters':['fact_sales.status = paid'],'time_dimension':'order_date','timezone':'UTC',    'owner':'growth-analytics','grain':'one distinct customer in the requested closed date window',    'description':'Customers with at least one paid order in [start_date,end_date].',  },  'period_retention.v1': {    'type':'metric','numerator':'customers active in both prior and current periods',    'denominator':'customers active in prior period','aggregation':'ratio_of_distinct_sets',    'filters':['fact_sales.status = paid'],'time_dimension':'order_date','timezone':'UTC',    'owner':'growth-analytics','grain':'one comparison of two closed date windows',    'zero_denominator':'NULL','description':'Retained / prior active; not retained / current active.',  },}SCHEMA = '''PRAGMA foreign_keys=ON;CREATE TABLE dim_customer_current(  customer_id TEXT PRIMARY KEY,  durable_customer_id TEXT NOT NULL,  customer_name TEXT NOT NULL,  geography_code TEXT NOT NULL,  lifecycle_status TEXT NOT NULL);CREATE TABLE dim_date(date_key TEXT PRIMARY KEY,year INTEGER,month INTEGER,day INTEGER,weekday_name TEXT);CREATE TABLE dim_product(product_id TEXT PRIMARY KEY,product_name TEXT NOT NULL,category TEXT NOT NULL);CREATE TABLE fact_sales(  order_id TEXT NOT NULL,line_no INTEGER NOT NULL,order_date TEXT NOT NULL,  product_id TEXT NOT NULL,customer_id TEXT NOT NULL,channel TEXT NOT NULL,  quantity INTEGER NOT NULL,amount_cents INTEGER NOT NULL,cost_cents INTEGER NOT NULL,  currency_code TEXT NOT NULL,status TEXT NOT NULL,source_seq INTEGER NOT NULL,  PRIMARY KEY(order_id,line_no),  FOREIGN KEY(order_date) REFERENCES dim_date(date_key),  FOREIGN KEY(product_id) REFERENCES dim_product(product_id),  FOREIGN KEY(customer_id) REFERENCES dim_customer_current(customer_id));CREATE TABLE agg_daily_product(  order_date TEXT NOT NULL, product_id TEXT NOT NULL,  line_count INTEGER NOT NULL, units INTEGER NOT NULL, gmv_cents INTEGER NOT NULL,  refreshed_through_seq INTEGER NOT NULL, PRIMARY KEY(order_date,product_id));CREATE TABLE metric_registry(  metric_name TEXT PRIMARY KEY,version INTEGER NOT NULL,status TEXT NOT NULL,  spec_json TEXT NOT NULL,spec_hash TEXT NOT NULL,owner TEXT NOT NULL,  deprecated_at TEXT,replacement_metric TEXT);CREATE TABLE semantic_lineage(  metric_name TEXT NOT NULL, upstream_object TEXT NOT NULL, upstream_field TEXT,  role TEXT NOT NULL, PRIMARY KEY(metric_name,upstream_object,upstream_field,role));CREATE TABLE principal_geography(  principal TEXT NOT NULL, geography_code TEXT NOT NULL,  PRIMARY KEY(principal,geography_code));'''def canonical_json(v):    return json.dumps(v, sort_keys=True, separators=(',',':'))def spec_hash(spec):    return hashlib.sha256(canonical_json(spec).encode()).hexdigest()def rows_hash(rows):    return hashlib.sha256(canonical_json(rows).encode()).hexdigest()def connect():    c = sqlite3.connect(DB)    c.executescript(SCHEMA)    c.executemany('INSERT INTO dim_customer_current VALUES (?,?,?,?,?)', CUSTOMERS)    c.executemany('INSERT INTO dim_date VALUES (?,?,?,?,?)', DATES)    c.executemany('INSERT INTO dim_product VALUES (?,?,?)', PRODUCTS)    c.executemany('INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?,?,?)', SALES)    c.execute('''INSERT INTO agg_daily_product      SELECT order_date,product_id,count(*),sum(quantity),sum(amount_cents),207      FROM fact_sales WHERE status='paid' AND currency_code='USD'      GROUP BY order_date,product_id''')    for name,spec in METRICS.items():        version=int(name.rsplit('.v',1)[1])        c.execute('INSERT INTO metric_registry VALUES (?,?,?,?,?,?,NULL,NULL)',                  (name,version,'active',canonical_json(spec),spec_hash(spec),spec['owner']))    # A deliberately ambiguous predecessor is retained as deprecated evidence, not rewritten in place.    legacy={'expression':'SUM(amount_cents)','time_semantics':'unspecified','currency_semantics':'unspecified','filters':'dashboard-owned'}    c.execute('INSERT INTO metric_registry VALUES (?,?,?,?,?,?,?,?)',              ('revenue_legacy.v0',0,'deprecated',canonical_json(legacy),spec_hash(legacy),'legacy-bi','2026-09-21T10:00:00Z','gross_revenue_usd.v1'))    lineage=[      ('gross_revenue_usd.v1','erp.order_lines','amount_cents','measure-source'),      ('gross_revenue_usd.v1','erp.orders','status','filter-source'),      ('gross_revenue_usd.v1','fact_sales','amount_cents','warehouse-measure'),      ('gross_revenue_usd.v1','fact_sales','order_date','time-dimension-key'),      ('active_customers.v1','erp.orders','customer_id','identity-source'),      ('active_customers.v1','fact_sales','customer_id','distinct-entity'),      ('period_retention.v1','fact_sales','customer_id','prior/current-set-membership'),      ('period_retention.v1','fact_sales','order_date','period-boundary'),    ]    c.executemany('INSERT INTO semantic_lineage VALUES (?,?,?,?)',lineage)    c.executemany('INSERT INTO principal_geography VALUES (?,?)',[('analyst_east','G-EAST'),('analyst_all','G-NORTH'),('analyst_all','G-EAST'),('analyst_all','G-SOUTH'),('analyst_all','__UNKNOWN__')])    c.commit(); return cdef controls(c):    r=c.execute('''SELECT count(*),count(distinct order_id),sum(quantity),sum(amount_cents),sum(cost_cents),sum(amount_cents-cost_cents)                   FROM fact_sales WHERE status='paid' AND currency_code='USD' ''').fetchone()    return {'lines':r[0],'orders':r[1],'units':r[2],'revenue_usd':r[3]//100,'cost_usd':r[4]//100,'profit_usd':r[5]//100}def security_clause(principal):    if principal is None: return '', []    return ''' JOIN dim_customer_current dc ON dc.customer_id=f.customer_id               JOIN principal_geography pg ON pg.geography_code=dc.geography_code AND pg.principal=? ''', [principal]def revenue(c,start='2026-09-18',end='2026-09-22',principal=None):    sec,params=security_clause(principal)    sql=f'''SELECT COALESCE(SUM(f.amount_cents),0) FROM fact_sales f {sec}            WHERE f.status='paid' AND f.currency_code='USD' AND f.order_date BETWEEN ? AND ?'''    return c.execute(sql,params+[start,end]).fetchone()[0]/100def active_customers(c,start,end,principal=None):    sec,params=security_clause(principal)    sql=f'''SELECT COUNT(DISTINCT f.customer_id) FROM fact_sales f {sec}            WHERE f.status='paid' AND f.order_date BETWEEN ? AND ?'''    return c.execute(sql,params+[start,end]).fetchone()[0]def retention(c,prior_start,prior_end,current_start,current_end,principal=None):    # RLS is applied to both sets before the ratio is calculated.    if principal is None:        prior_sql="SELECT DISTINCT customer_id FROM fact_sales WHERE status='paid' AND order_date BETWEEN ? AND ?"        current_sql="SELECT DISTINCT customer_id FROM fact_sales WHERE status='paid' AND order_date BETWEEN ? AND ?"        params=[prior_start,prior_end,current_start,current_end]    else:        set_sql='''SELECT DISTINCT f.customer_id FROM fact_sales f                   JOIN dim_customer_current dc ON dc.customer_id=f.customer_id                   JOIN principal_geography pg ON pg.geography_code=dc.geography_code AND pg.principal=?                   WHERE f.status='paid' AND f.order_date BETWEEN ? AND ?'''        prior_sql=set_sql; current_sql=set_sql        params=[principal,prior_start,prior_end,principal,current_start,current_end]    q=f'''WITH prior AS ({prior_sql}), current AS ({current_sql}),         counts AS (SELECT (SELECT count(*) FROM prior) prior_n,                           (SELECT count(*) FROM current) current_n,                           (SELECT count(*) FROM prior p JOIN current c USING(customer_id)) retained_n)         SELECT prior_n,current_n,retained_n,                CASE WHEN prior_n=0 THEN NULL ELSE retained_n*1.0/prior_n END AS retention_rate         FROM counts'''    return c.execute(q,params).fetchone()def compile_sql(metric_name, principal=None):    # Small didactic compiler: the contract owns semantics; clients receive SQL.    if metric_name=='gross_revenue_usd.v1':        sec="" if principal is None else "JOIN dim_customer_current dc ON dc.customer_id=f.customer_id JOIN principal_geography pg ON pg.geography_code=dc.geography_code AND pg.principal=:principal"        return f"SELECT SUM(f.amount_cents)/100.0 AS gross_revenue_usd FROM fact_sales f {sec} WHERE f.status='paid' AND f.currency_code='USD' AND f.order_date BETWEEN :start_date AND :end_date"    if metric_name=='active_customers.v1':        sec="" if principal is None else "JOIN dim_customer_current dc ON dc.customer_id=f.customer_id JOIN principal_geography pg ON pg.geography_code=dc.geography_code AND pg.principal=:principal"        return f"SELECT COUNT(DISTINCT f.customer_id) AS active_customers FROM fact_sales f {sec} WHERE f.status='paid' AND f.order_date BETWEEN :start_date AND :end_date"    raise KeyError(metric_name)def main():    c=connect()    ctl=controls(c); assert ctl==EXPECTED_CONTROLS,ctl    base_revenue=revenue(c); assert base_revenue==820    current_active=active_customers(c,'2026-09-20','2026-09-22'); assert current_active==4    prior_active=active_customers(c,'2026-09-18','2026-09-19'); assert prior_active==2    ret=retention(c,'2026-09-18','2026-09-19','2026-09-20','2026-09-22')    assert ret==(2,4,1,0.5),ret    zero=retention(c,'2026-09-01','2026-09-02','2026-09-20','2026-09-22')    assert zero[0]==0 and zero[3] is None    east_revenue=revenue(c,principal='analyst_east'); assert east_revenue==395    east_ret=retention(c,'2026-09-18','2026-09-19','2026-09-20','2026-09-22','analyst_east')    assert east_ret==(1,2,1,1.0),east_ret    agg_revenue=c.execute('SELECT sum(gmv_cents)/100.0 FROM agg_daily_product').fetchone()[0]    assert agg_revenue==base_revenue    # Controlled dashboard drift.    hidden_cutoff=c.execute("SELECT sum(amount_cents)/100.0 FROM fact_sales WHERE status='paid' AND order_date<='2026-09-21'").fetchone()[0]    row_count_active=c.execute("SELECT count(*) FROM fact_sales WHERE status='paid' AND order_date BETWEEN '2026-09-20' AND '2026-09-22'").fetchone()[0]    wrong_retained_over_current=ret[2]/ret[1]    assert hidden_cutoff==690 and row_count_active==6 and wrong_retained_over_current==0.25    # Mixed currency is blocked rather than silently summed.    c.execute("CREATE TEMP TABLE currency_probe AS SELECT * FROM fact_sales")    c.execute("INSERT INTO currency_probe VALUES ('T-EUR',1,'2026-09-22','P100','C001','web',1,1000,500,'EUR','paid',999)")    currencies=[x[0] for x in c.execute("SELECT DISTINCT currency_code FROM currency_probe WHERE status='paid'")]    currency_gate='BLOCK' if len(currencies)>1 else 'PASS'    assert currency_gate=='BLOCK'    registry=[dict(zip(['metric_name','version','status','spec_hash','owner','deprecated_at','replacement_metric'],r)) for r in c.execute('SELECT metric_name,version,status,spec_hash,owner,deprecated_at,replacement_metric FROM metric_registry ORDER BY metric_name')]    lineage=[dict(zip(['metric_name','upstream_object','upstream_field','role'],r)) for r in c.execute('SELECT metric_name,upstream_object,upstream_field,role FROM semantic_lineage ORDER BY metric_name,upstream_object,upstream_field')]    compiled={      'revenue_unrestricted':compile_sql('gross_revenue_usd.v1'),      'revenue_rls':compile_sql('gross_revenue_usd.v1','analyst_east'),      'active_unrestricted':compile_sql('active_customers.v1')    }    golden={      'gross_revenue_usd.v1':820.0,      'active_customers.v1[2026-09-20..2026-09-22]':4,      'period_retention.v1[2026-09-18..19 -> 2026-09-20..22]':0.5,      'gross_revenue_usd.v1[analyst_east]':395.0,      'period_retention.v1[analyst_east]':1.0,    }    report={      'runtime':{'python':sys.version.split()[0],'sqlite':sqlite3.sqlite_version},      'controls':ctl,      'metric_specs':METRICS,      'metric_hashes':{k:spec_hash(v) for k,v in METRICS.items()},      'golden_results':golden,      'retention_counts':{'prior':ret[0],'current':ret[1],'retained':ret[2]},      'zero_denominator_retention':zero[3],      'acceleration_crosscheck':{'base_revenue':base_revenue,'aggregate_revenue':agg_revenue,'equal':base_revenue==agg_revenue},      'controlled_drift':{        'hidden_end_date_revenue':hidden_cutoff,'governed_revenue':base_revenue,        'row_count_mislabeled_active_customers':row_count_active,'governed_active_customers':current_active,        'retained_over_current_wrong':wrong_retained_over_current,'retained_over_prior_governed':ret[3]      },      'security':{'analyst_east_revenue':east_revenue,'analyst_east_retention':east_ret[3]},      'currency_probe':{'currencies':currencies,'decision':currency_gate},      'registry':registry,'lineage':lineage,'compiled_sql':compiled,      'fact_hash':rows_hash(c.execute('SELECT * FROM fact_sales ORDER BY order_id,line_no').fetchall()),      'registry_hash':rows_hash(c.execute('SELECT metric_name,version,status,spec_hash,owner,deprecated_at,replacement_metric FROM metric_registry ORDER BY metric_name').fetchall())    }    (OUT/'metric_specs.json').write_text(json.dumps(METRICS,indent=2,sort_keys=True),encoding='utf-8')    (OUT/'report.json').write_text(json.dumps(report,indent=2,sort_keys=True),encoding='utf-8')    print(json.dumps(report,indent=2,sort_keys=True))    c.close()if __name__=='__main__': main()

gross_revenue_usd.v1, all dates:                   820.00 USDactive_customers.v1, 2026-09-20..2026-09-22:        4period_retention.v1, Sep18-19 -> Sep20-22:           0.50 (50%)  prior active customers:                             2  current active customers:                           4  retained customers:                                 1gross_revenue_usd.v1, analyst_east:                 395.00 USDperiod_retention.v1, analyst_east:                    1.00 (100%)zero-prior-population retention:                      NULLatomic revenue vs Chapter 19 accelerator:             820.00 == 820.00

governed gross revenue:                         820.00 USDwrong dashboard with hidden end_date=2026-09-21:    690.00 USDgoverned active customers (Sep20..Sep22):              4wrong dashboard COUNT(*) mislabeled as customers:       6governed retention = retained / prior active:          50%wrong dashboard = retained / current active:            25%