Setting the file. One moment.
Breakdown Attribution · Investigate Metric · PostHog/skills · Skills Docs
Back to the top of the page 191
Feature Flags Node.js
11
Consuming Endpoints From Client Code
scripts/breakdown_attribution.py
scripts/ breakdown_attribution.py
Python · 168 lines · 6 KB
16
Optional env:
17 WINDOW=N How many trailing intervals (days/hours/etc., depending
18 on the input's interval) to treat as the anomaly window.
19 Default 7. The preceding N intervals are the baseline.
20 TOP=N Show only the top N segments (default 10).
21 """
22
23 from __future__ import annotations
24
25 import json
26 import os
27 import sys
28
29
30 def load_input () -> dict :
31 if len (sys.argv) > 1 :
32 with open (sys.argv[ 1 ]) as f:
33 raw = f.read()
34 else :
35 raw = sys.stdin.read()
36 parsed = json.loads(raw)
37 if isinstance (parsed, list ) and parsed and parsed[ 0 ].get( "type" ) == "text" :
38 parsed = json.loads(parsed[ 0 ][ "text" ])
39 return parsed
40
41
42 def fmt (v: float , signed: bool = False ) -> str :
43 if signed:
44 prefix = "+" if v >= 0 else "-"
45 else :
46 prefix = "" if v >= 0 else "-"
47 a = abs (v)
48 if a >= 1_000_000 :
49 return f " { prefix }{ a / 1_000_000 :.2f} M"
50 if a >= 1_000 :
51 return f " { prefix }{ a / 1_000 :.1f} K"
52 return f " { prefix }{ a :,.0f} "
53
54
55 def fmt_pct (v: float ) -> str :
56 # `v == v` is False only when v is NaN.
57 return f " { v :+.1f} %" if v == v else "n/a"
58
59
60 def main () -> int :
61 window = int (os.environ.get( "WINDOW" , "7" ))
62 top = int (os.environ.get( "TOP" , "10" ))
63
64 payload = load_input()
65 results = payload.get( "results" ) or payload.get( "result" ) or []
66 if not results:
67 raise SystemExit ( "No results in payload — is this a breakdown trends response?" )
68
69 rows = []
70 total_anomaly = 0.0
71 total_baseline = 0.0
72
73 for series in results:
74 data = series.get( "data" ) or []
75 if len (data) < 2 * window:
76 print (
77 f "warn: series ' { series.get( 'breakdown_value' , series.get( 'label' )) } ' "
78 f "has { len (data) } points but window*2= { 2 * window } — "
79 "skipping (extend dateRange)." ,
80 file = sys.stderr,
81 )
82 continue
83 baseline = sum (data[ - 2 * window : - window])
84 current = sum (data[ - window:])
85 delta = current - baseline
86 total_anomaly += current
87 total_baseline += baseline
88
89 seg = series.get( "breakdown_value" )
90 if seg is None or seg == "" :
91 seg = series.get( "label" , "(none)" )
92 if isinstance (seg, list ):
93 seg = " / " .join( str (x) for x in seg)
94 rows.append({
95 "segment" : str (seg),
96 "baseline" : baseline,
97 "current" : current,
98 "delta" : delta,
99 "pct" : (delta / baseline * 100 ) if baseline else float ( "nan" ),
100 })
101
102 if not rows:
103 raise SystemExit (
104 "No usable series — every breakdown had fewer than 2 windows of data. "
105 "Run with a wider dateRange."
106 )
107
108 rows.sort( key =lambda r: abs (r[ "delta" ]), reverse = True )
109 total_delta = total_anomaly - total_baseline
110
111 print ( f "# Breakdown attribution — last { window } intervals vs preceding { window } intervals" )
112 print ()
113 print ( f "Aggregate: { fmt(total_baseline) } → { fmt(total_anomaly) } ( { fmt(total_delta, signed = True ) } , "
114 f " { fmt_pct((total_delta / total_baseline * 100 ) if total_baseline else float ( 'nan' )) } )" )
115 print ()
116 print ( "Segments ranked by **absolute** delta contribution:" )
117 print ()
118 print ( "| Segment | Baseline | Current | Δ | Δ% | Share of total Δ |" )
119 print ( "| --- | ---: | ---: | ---: | ---: | ---: |" )
120
121 for r in rows[:top]:
122 share = (r[ "delta" ] / total_delta * 100 ) if total_delta else float ( "nan" )
123 print (
124 f "| { r[ 'segment' ] } "
125 f "| { fmt(r[ 'baseline' ]) } "
126 f "| { fmt(r[ 'current' ]) } "
127 f "| { fmt(r[ 'delta' ], signed = True ) } "
128 f "| { fmt_pct(r[ 'pct' ]) } "
129 f "| { fmt_pct(share) } |"
130 )
131
132 print ()
133 # If the aggregate barely moved but segments did, segments are offsetting.
134 # That's a different diagnostic than "one segment absorbs the delta".
135 aggregate_pct = (total_delta / total_baseline * 100 ) if total_baseline else 0
136 largest_segment_move = max ( abs (r[ "delta" ]) for r in rows)
137 aggregate_is_quiet = abs (aggregate_pct) < 5 and largest_segment_move > abs (total_delta) * 2
138
139 if aggregate_is_quiet:
140 print (
141 "**Aggregate barely moved but individual segments did — segments are "
142 "offsetting each other. Investigate the largest movers separately rather "
143 "than as a 'share of total delta'.**"
144 )
145 return 0
146
147 top_row = rows[ 0 ]
148 top_share = (top_row[ "delta" ] / total_delta * 100 ) if total_delta else float ( "nan" )
149 if total_delta and abs (top_share) >= 50 :
150 print (
151 f "**Top segment ' { top_row[ 'segment' ] } ' absorbs "
152 f " { abs (top_share) :.0f} % of the aggregate delta — strong segment signal.**"
153 )
154 elif total_delta and sum ( abs (r[ "delta" ]) for r in rows[: 3 ]) / abs (total_delta) >= 0.7 :
155 print (
156 "**Top 3 segments account for ≥70 % o f the delta — investigate what they share.**"
157 )
158 else :
159 print (
160 "**No single segment dominates — the cause is likely system-wide "
161 "(deploy, tracking, infra) rather than segment-specific.**"
162 )
163
164 return 0
165
166
167 if __name__ == "__main__" :
168 sys.exit(main())