Design a Marketing/Finance/Operations Mart Strategy that Reuses Conformed Data Without Blocking Domain Delivery

governed warehouse + semantic contracts        |        +--> finance mart     (revenue, cost, profit)        +--> marketing mart   (customer/segment/geography + governed revenue)        +--> operations mart  (orders, units + operational extensions)        |        +--> self-service sandboxes                 |                 +-- promotion gate --> certified domain productshared concepts: inherited + versioneddomain-only concepts: domain-owned + documentedcertification: evidence-based, not schema-name-based

from __future__ import annotationsimport hashlib, json, sqlite3, sysfrom pathlib import PathOUT=Path('/mnt/data/atlasmart_ch21_lab'); OUT.mkdir(parents=True,exist_ok=True)DB=OUT/'atlasmart_ch21.sqlite'if DB.exists(): DB.unlink()SALES=[('O1001',1,'2026-09-18','P100','C001','web',2,10000,6000,'USD','paid'),('O1001',2,'2026-09-18','P200','C001','web',1,2500,1500,'USD','paid'),('O1002',1,'2026-09-18','P300','C002','mobile',1,19000,12000,'USD','paid'),('O1003',1,'2026-09-19','P400','C001','web',1,10000,6500,'USD','paid'),('O1005',1,'2026-09-20','P100','C004','mobile',1,5000,3000,'USD','paid'),('O1005',2,'2026-09-20','P400','C004','mobile',1,10000,6500,'USD','paid'),('O1007',1,'2026-09-21','P300','C003','sales',1,7500,4500,'USD','paid'),('O1008',1,'2026-09-21','P200','C002','mobile',2,5000,2500,'USD','paid'),('O1009',1,'2026-09-22','P100','C005','web',1,5000,2500,'USD','paid'),('O1010',1,'2026-09-22','P100','C003','sales',1,8000,4500,'USD','paid')]CUSTOMERS=[('C001','D-CUST-001','Ada Retail','Mid-Market','G-NORTH','active'),('C002','D-CUST-002','Ben Home','Consumer','G-EAST','active'),('C003','D-CUST-003','Cyra Labs','Enterprise','G-EAST','active'),('C004','D-CUST-004','Dara Studio','Consumer','G-SOUTH','inactive'),('C005','D-CUST-005','__INFERRED__','__UNKNOWN__','__UNKNOWN__','active')]PRODUCTS=[('P100','Keyboard','Accessories'),('P200','Mouse','Accessories'),('P300','Monitor','Displays'),('P400','Dock','Accessories')]METRIC_HASHES={'gross_revenue_usd.v1':'6be77e930e087257e5756fc2c9ccddd4293fed6a457d3a8c72badef65dad5e8a','active_customers.v1':'1fb6a9bb4c820d321fe786b0809124fa279182c0f6a28e81db1194f37ada1eb0','period_retention.v1':'66e7840903abf5f871d39f72c35935d3d98cbadb8496480342f781777da3a962'}EXPECTED={'lines':10,'orders':8,'units':12,'revenue_usd':820,'cost_usd':495,'profit_usd':325}def h(v): return hashlib.sha256(json.dumps(v,sort_keys=True,separators=(',',':')).encode()).hexdigest()def setup(): c=sqlite3.connect(DB) c.executescript(''' PRAGMA foreign_keys=ON; CREATE TABLE fact_sales(order_id TEXT,line_no INT,order_date TEXT,product_id TEXT,customer_id TEXT,channel TEXT,quantity INT,amount_cents INT,cost_cents INT,currency_code TEXT,status TEXT,PRIMARY KEY(order_id,line_no)); CREATE TABLE dim_customer(customer_id TEXT PRIMARY KEY,durable_customer_id TEXT,customer_name TEXT,segment TEXT,geography_code TEXT,lifecycle_status TEXT); CREATE TABLE dim_product(product_id TEXT PRIMARY KEY,product_name TEXT,category TEXT); CREATE TABLE metric_registry(metric_name TEXT PRIMARY KEY,spec_hash TEXT NOT NULL,status TEXT NOT NULL,owner TEXT NOT NULL); CREATE TABLE mart_registry(mart_name TEXT PRIMARY KEY,domain TEXT,owner TEXT,status TEXT,certified_at TEXT,access_class TEXT,contract_hash TEXT); CREATE TABLE mart_dependency(mart_name TEXT,upstream_object TEXT,dependency_type TEXT,PRIMARY KEY(mart_name,upstream_object)); CREATE TABLE promotion_check(run_id TEXT,mart_name TEXT,check_name TEXT,result TEXT,evidence TEXT,PRIMARY KEY(run_id,mart_name,check_name)); ''') c.executemany('INSERT INTO fact_sales VALUES (?,?,?,?,?,?,?,?,?,?,?)',SALES) c.executemany('INSERT INTO dim_customer VALUES (?,?,?,?,?,?)',CUSTOMERS) c.executemany('INSERT INTO dim_product VALUES (?,?,?)',PRODUCTS) for n,sh in METRIC_HASHES.items(): c.execute('INSERT INTO metric_registry VALUES (?,?,?,?)',(n,sh,'active','enterprise-semantic')) 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 build_dependent_marts(c): c.executescript(''' DROP VIEW IF EXISTS mart_finance_daily; CREATE VIEW mart_finance_daily AS SELECT order_date, SUM(amount_cents)/100.0 AS gross_revenue_usd,        SUM(cost_cents)/100.0 AS cost_usd, SUM(amount_cents-cost_cents)/100.0 AS gross_profit_usd FROM fact_sales WHERE status='paid' AND currency_code='USD' GROUP BY order_date; DROP VIEW IF EXISTS mart_operations_daily; CREATE VIEW mart_operations_daily AS SELECT order_date, COUNT(DISTINCT order_id) AS orders, SUM(quantity) AS units FROM fact_sales WHERE status='paid' GROUP BY order_date; DROP VIEW IF EXISTS mart_marketing_customer_daily; CREATE VIEW mart_marketing_customer_daily AS SELECT f.order_date,f.customer_id,c.segment,c.geography_code,        SUM(f.amount_cents)/100.0 AS governed_revenue_usd FROM fact_sales f JOIN dim_customer c USING(customer_id) WHERE f.status='paid' AND f.currency_code='USD' GROUP BY f.order_date,f.customer_id,c.segment,c.geography_code; DROP VIEW IF EXISTS mart_marketing_wide; CREATE VIEW mart_marketing_wide AS SELECT f.order_date,f.order_id,f.line_no,f.customer_id,c.customer_name,c.segment,c.geography_code,        f.product_id,p.product_name,p.category,f.channel,f.quantity,f.amount_cents/100.0 AS governed_revenue_usd FROM fact_sales f JOIN dim_customer c USING(customer_id) JOIN dim_product p USING(product_id) WHERE f.status='paid' AND f.currency_code='USD'; ''') for name,domain,owner,access in [ ('mart_finance_daily','finance','finance-analytics','finance-certified'), ('mart_operations_daily','operations','operations-analytics','operations-certified'), ('mart_marketing_customer_daily','marketing','growth-analytics','marketing-certified'), ('mart_marketing_wide','marketing','growth-analytics','marketing-certified')]:  deps=['fact_sales','gross_revenue_usd.v1'] if 'finance' in name else ['fact_sales']  if 'marketing' in name: deps=['fact_sales','dim_customer','gross_revenue_usd.v1'] + (['dim_product'] if 'wide' in name else [])  contract={'mart':name,'domain':domain,'owner':owner,'dependencies':deps,'access':access}  c.execute('INSERT OR REPLACE INTO mart_registry VALUES (?,?,?,?,?,?,?)',(name,domain,owner,'certified','2026-09-21T11:00:00Z',access,h(contract)))  for d in deps: c.execute('INSERT OR REPLACE INTO mart_dependency VALUES (?,?,?)',(name,d,'governed')) c.commit()def make_bad_sandbox(c): c.executescript('''DROP TABLE IF EXISTS sandbox_marketing_revenue; CREATE TABLE sandbox_marketing_revenue AS SELECT order_date, SUM(amount_cents)/100.0 AS revenue FROM fact_sales WHERE status='paid' AND currency_code='USD' AND channel <> 'sales' GROUP BY order_date;''')def scan_drift(c): governed=c.execute("SELECT SUM(amount_cents)/100.0 FROM fact_sales WHERE status='paid' AND currency_code='USD'").fetchone()[0] bad=c.execute('SELECT SUM(revenue) FROM sandbox_marketing_revenue').fetchone()[0] return {'governed_revenue_usd':governed,'sandbox_revenue_usd':bad,'delta_usd':bad-governed,'status':'DRIFT' if bad!=governed else 'PASS'}def promote(c,run_id,mart_name,repaired=False): checks=[] if repaired:  observed=c.execute('SELECT SUM(governed_revenue_usd) FROM sandbox_marketing_revenue_fixed').fetchone()[0] else: observed=c.execute('SELECT SUM(revenue) FROM sandbox_marketing_revenue').fetchone()[0] checks.append(('metric_reconciliation','PASS' if observed==820 else 'FAIL',f'observed={observed:.2f}; expected=820.00')) checks.append(('metric_version','PASS' if repaired else 'FAIL',f"required gross_revenue_usd.v1 hash={METRIC_HASHES['gross_revenue_usd.v1']}")) checks.append(('conformed_customer','PASS' if repaired else 'FAIL','must consume governed dim_customer rather than team-local customer semantics')) checks.append(('owner','PASS','owner=growth-analytics')) checks.append(('access_class','PASS','access=marketing-certified')) for n,res,ev in checks: c.execute('INSERT OR REPLACE INTO promotion_check VALUES (?,?,?,?,?)',(run_id,mart_name,n,res,ev)) decision='CERTIFY' if all(x[1]=='PASS' for x in checks) else 'BLOCK' return decision,checksdef repair_sandbox(c): c.executescript('''DROP VIEW IF EXISTS sandbox_marketing_revenue_fixed; CREATE VIEW sandbox_marketing_revenue_fixed AS SELECT f.order_date,c.segment,c.geography_code,SUM(f.amount_cents)/100.0 AS governed_revenue_usd FROM fact_sales f JOIN dim_customer c USING(customer_id) WHERE f.status='paid' AND f.currency_code='USD' GROUP BY f.order_date,c.segment,c.geography_code;''') contract={'mart':'marketing_revenue_v1','dependencies':['fact_sales','dim_customer','gross_revenue_usd.v1'],'metric_hash':METRIC_HASHES['gross_revenue_usd.v1'],'owner':'growth-analytics','access':'marketing-certified'} c.execute('INSERT OR REPLACE INTO mart_registry VALUES (?,?,?,?,?,?,?)',('marketing_revenue_v1','marketing','growth-analytics','certified','2026-09-21T11:30:00Z','marketing-certified',h(contract))) for d in contract['dependencies']: c.execute('INSERT OR REPLACE INTO mart_dependency VALUES (?,?,?)',('marketing_revenue_v1',d,'governed')) c.commit(); return contractdef main(): c=setup(); ctl=controls(c); assert ctl==EXPECTED,ctl build_dependent_marts(c) fin=c.execute('SELECT SUM(gross_revenue_usd),SUM(cost_usd),SUM(gross_profit_usd) FROM mart_finance_daily').fetchone(); assert fin==(820.0,495.0,325.0) ops=c.execute('SELECT SUM(orders),SUM(units) FROM mart_operations_daily').fetchone(); assert ops==(8,12) mkt=c.execute('SELECT SUM(governed_revenue_usd),COUNT(DISTINCT customer_id) FROM mart_marketing_customer_daily').fetchone(); assert mkt==(820.0,5) wide=c.execute('SELECT COUNT(*),SUM(governed_revenue_usd) FROM mart_marketing_wide').fetchone(); assert wide==(10,820.0) make_bad_sandbox(c); drift=scan_drift(c); assert drift['sandbox_revenue_usd']==665.0 and drift['status']=='DRIFT',drift bad_dec,bad_checks=promote(c,'PROMO-001','sandbox_marketing_revenue',False); assert bad_dec=='BLOCK' contract=repair_sandbox(c); fixed=c.execute('SELECT SUM(governed_revenue_usd) FROM sandbox_marketing_revenue_fixed').fetchone()[0]; assert fixed==820.0 good_dec,good_checks=promote(c,'PROMO-002','marketing_revenue_v1',True); assert good_dec=='CERTIFY' deps=[dict(zip(['mart_name','upstream_object','dependency_type'],r)) for r in c.execute('SELECT * FROM mart_dependency ORDER BY mart_name,upstream_object')] regs=[dict(zip(['mart_name','domain','owner','status','certified_at','access_class','contract_hash'],r)) for r in c.execute('SELECT * FROM mart_registry ORDER BY mart_name')] checks=[dict(zip(['run_id','mart_name','check_name','result','evidence'],r)) for r in c.execute('SELECT * FROM promotion_check ORDER BY run_id,check_name')] duplicate_scan={  'independent_business_definitions':['sandbox_marketing_revenue.revenue excludes channel=sales but uses generic revenue name'],  'governed_metric_reuse':sorted({r[1] for r in c.execute("SELECT mart_name,upstream_object FROM mart_dependency WHERE upstream_object LIKE '%.v1'")}),  'status':'DRIFT_DETECTED_AND_REPAIRED'} report={'runtime':{'python':sys.version.split()[0],'sqlite':sqlite3.sqlite_version},'controls':ctl,  'domain_mart_results':{'finance':{'revenue_usd':fin[0],'cost_usd':fin[1],'profit_usd':fin[2]},'operations':{'orders':ops[0],'units':ops[1]},'marketing':{'revenue_usd':mkt[0],'customers_all_time':mkt[1]},'wide':{'rows':wide[0],'revenue_usd':wide[1]}},  'drift':drift,'bad_promotion':{'decision':bad_dec,'checks':bad_checks},'repaired_promotion':{'decision':good_dec,'checks':good_checks},  'repaired_contract':contract,'mart_registry':regs,'dependencies':deps,'promotion_evidence':checks,'duplicate_definition_scan':duplicate_scan,  'warehouse_hash':h(SALES),'metric_hashes':METRIC_HASHES} (OUT/'report.json').write_text(json.dumps(report,indent=2,sort_keys=True),encoding='utf-8') (OUT/'dependency_map.json').write_text(json.dumps(deps,indent=2),encoding='utf-8') (OUT/'promotion_checklist.json').write_text(json.dumps(checks,indent=2),encoding='utf-8') print(json.dumps(report,indent=2,sort_keys=True)); c.close()if __name__=='__main__': main()

Finance certified mart  revenue_usd: 820.00  cost_usd:    495.00  profit_usd:  325.00Operations certified mart  orders:      8  units:       12Marketing certified mart  revenue_usd: 820.00  customers:   5Certified wide delivery surface  rows:        10  revenue_usd: 820.00

governed gross_revenue_usd.v1:        820.00 USDindependent marketing "revenue":        665.00 USDdelta:                                 -155.00 USDreason: hidden predicate channel <> 'sales'certification decision before repair:   BLOCKcertification decision after repair:    CERTIFY

atlasmart_ch21_lab/  atlasmart_ch21.sqlite  report.json  dependency_map.json  promotion_checklist.jsonreport.json  runtime + warehouse controls  domain mart results  drift evidence  blocked/successful promotion results  registry + contract hashes  dependency graph  duplicate-definition scan

SELECT order_date,gross_revenue_usd,cost_usd,gross_profit_usdFROM mart_finance_dailyORDER BY order_date;

SELECT order_date,orders,unitsFROM mart_operations_dailyORDER BY order_date;

SELECT segment,geography_code,SUM(governed_revenue_usd) AS revenue_usdFROM mart_marketing_customer_dailyGROUP BY segment,geography_codeORDER BY segment,geography_code;