Janady13/missionmatch / internal/modules/eeo/fairness.go

1package eeo 2 3import ( 4 "context" 5 "time" 6 7 "github.com/jackc/pgx/v5" 8 "github.com/shopspring/decimal" 9) 10 11// driftDelta is the fixed drop in a group's four fifths impact ratio, snapshot over snapshot, that 12// raises a drift alert even while the ratio is still at or above 0.80. A best in world AEDT tool warns 13// BEFORE a group crosses the threshold, so a 0.05 fall between two consecutive snapshots is surfaced. 14var driftDelta = decimal.RequireFromString("0.05") 15 16// SnapshotSummary is the group level aggregate result of one SnapshotAdverseImpact run for one org. It 17// carries NO per person data: only the count of rows written and how many groups fell below the four 18// fifths threshold, so a scheduler can log a concise line with no PII. 19type SnapshotSummary struct { 20 OrganizationID string 21 GroupsWritten int 22 BelowThreshold int 23 ModelVersion string 24 SnapshotAtRFC string 25} 26 27// FairnessPoint is one demographic group bucket's impact ratio at one snapshot in the time series. The 28// ratio is a decimal string exactly as the AdverseImpact report labels it: the highest rate reference 29// group passes at 1.0 against itself, and "n/a" marks an unknown group or a cut with fewer than two known 30// groups (never compared, never flagged). 31type FairnessPoint struct { 32 SnapshotAt string `json:"snapshot_at"` 33 SelectionRate string `json:"selection_rate"` 34 ImpactRatio string `json:"impact_ratio"` 35 Selected int `json:"selected"` 36 Total int `json:"total"` 37 ModelVersion string `json:"model_version"` 38} 39 40// FairnessSeries is the continuous time series for one (decision_context, group_label) bucket, with an 41// alert flag. Alert is true when the LATEST snapshot is below 0.80, OR when the impact ratio dropped by 42// more than driftDelta from the prior snapshot (drift toward disparate impact). This is GROUP LEVEL 43// aggregate data; it is never per person and never an input to matching. 44type FairnessSeries struct { 45 DecisionContext string `json:"decision_context"` 46 GroupLabel string `json:"group_label"` 47 Points []FairnessPoint `json:"points"` 48 Alert bool `json:"alert"` 49 AlertReason string `json:"alert_reason"` 50} 51 52// SnapshotAdverseImpact computes the current adverse impact report for the caller's org and APPENDS one 53// fairness_snapshots row per demographic group bucket through the compliance pool, stamping the model 54// version in force at snapshot time. It is append only: it never updates or deletes a prior snapshot. 55// Run on a schedule, it builds the continuous four fifths time series the drift alert reads. 56// 57// The whole read plus write runs inside one tenant transaction, so RLS scopes both the AdverseImpact 58// join and the snapshot insert to the caller's org, and a partial run rolls back cleanly. 59func (s *Service) SnapshotAdverseImpact(ctx context.Context, now time.Time) (SnapshotSummary, error) { 60 if !s.compliance.Ready() { 61 return SnapshotSummary{}, ErrNoCompliance 62 } 63 rep, err := s.AdverseImpact(ctx, now) 64 if err != nil { 65 return SnapshotSummary{}, err 66 } 67 68 snapAt := now.UTC() 69 summary := SnapshotSummary{SnapshotAtRFC: snapAt.Format(time.RFC3339)} 70 rows := snapshotRowsFromReport(rep) 71 72 err = s.compliance.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 73 model, e := activeModelVersion(ctx, tx) 74 if e != nil { 75 return e 76 } 77 summary.ModelVersion = model 78 for _, r := range rows { 79 if e := insertSnapshot(ctx, tx, snapAt, model, r); e != nil { 80 return e 81 } 82 summary.GroupsWritten++ 83 if r.belowThreshold { 84 summary.BelowThreshold++ 85 } 86 } 87 return nil 88 }) 89 if err != nil { 90 return SnapshotSummary{}, err 91 } 92 return summary, nil 93} 94 95// snapshotRow is one group bucket flattened out of the report, ready to append. The rate and ratio are 96// decimals (never float64) so the numeric column round trips bit stable. 97type snapshotRow struct { 98 decisionContext string 99 groupLabel string 100 selected int 101 total int 102 selectionRate decimal.Decimal 103 impactRatio *decimal.Decimal // nil only for an n/a row, an unknown group or a cut with fewer than two known groups; the reference group passes at 1.0 104 belowThreshold bool 105} 106 107// snapshotRowsFromReport flattens every ImpactRow across both stages and all three cuts into append 108// ready rows. decision_context is the stage plus the cut, so the series for one group is unambiguous and 109// mirrors exactly what AdverseImpact reports. The rate and ratio strings the report already computed are 110// parsed back into decimal so the snapshot is provably the same number the report shows. 111func snapshotRowsFromReport(rep Report) []snapshotRow { 112 var out []snapshotRow 113 add := func(stage, cut string, imp []ImpactRow) { 114 ctxLabel := stage + " :: " + cut 115 for _, r := range imp { 116 out = append(out, snapshotRowFromImpact(ctxLabel, r)) 117 } 118 } 119 add(rep.Selection.Stage, "by sex", rep.Selection.ByGender) 120 add(rep.Selection.Stage, "by race or ethnicity", rep.Selection.ByRace) 121 add(rep.Selection.Stage, "intersectional", rep.Selection.Intersectional) 122 add(rep.Screening.Stage, "by sex", rep.Screening.ByGender) 123 add(rep.Screening.Stage, "by race or ethnicity", rep.Screening.ByRace) 124 add(rep.Screening.Stage, "intersectional", rep.Screening.Intersectional) 125 return out 126} 127 128// snapshotRowFromImpact converts one report ImpactRow into an append ready snapshot row, recomputing 129// below_threshold from the parsed ratio so it matches the report Flagged flag exactly. 130func snapshotRowFromImpact(decisionContext string, r ImpactRow) snapshotRow { 131 row := snapshotRow{ 132 decisionContext: decisionContext, 133 groupLabel: r.Group, 134 selected: r.Selected, 135 total: r.Considered, 136 selectionRate: decimalOrZero(r.Rate), 137 } 138 if r.ImpactRatio != "n/a" { 139 ratio := decimalOrZero(r.ImpactRatio) 140 row.impactRatio = &ratio 141 row.belowThreshold = ratio.LessThan(fourFifths) 142 } 143 return row 144} 145 146// decimalOrZero parses a fixed decimal string, returning zero on any unparsable value. The report only 147// ever emits clean fixed strings here, so this is defensive, not a real path. 148func decimalOrZero(s string) decimal.Decimal { 149 d, err := decimal.NewFromString(s) 150 if err != nil { 151 return decimal.Zero 152 } 153 return d 154} 155 156// FairnessTrend reads the appended snapshots for the caller's org since the given instant and returns, 157// per (decision_context, group_label) bucket, the ordered series of impact ratios and an alert flag. The 158// alert fires when the latest impact ratio is below 0.80, OR when it dropped by more than driftDelta from 159// the prior snapshot. Decimals throughout, never float64. 160func (s *Service) FairnessTrend(ctx context.Context, since time.Time) ([]FairnessSeries, error) { 161 if !s.compliance.Ready() { 162 return nil, ErrNoCompliance 163 } 164 var series []FairnessSeries 165 err := s.compliance.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 166 var e error 167 series, e = readTrend(ctx, tx, since) 168 return e 169 }) 170 if err != nil { 171 return nil, err 172 } 173 return series, nil 174} 175 176// readTrend loads the ordered snapshots, groups them into per bucket series, and computes the alert for 177// each. The SQL orders by bucket then time so the points within a series are already chronological. 178func readTrend(ctx context.Context, tx pgx.Tx, since time.Time) ([]FairnessSeries, error) { 179 rows, err := tx.Query(ctx, trendQuery, since.UTC()) 180 if err != nil { 181 return nil, err 182 } 183 defer rows.Close() 184 185 type key struct{ ctx, group string } 186 order := make([]key, 0) 187 byBucket := make(map[key]*FairnessSeries) 188 for rows.Next() { 189 var ( 190 decisionContext, groupLabel, modelVersion string 191 snapshotAt time.Time 192 selected, total int 193 selectionRate decimal.Decimal 194 impactRatio *decimal.Decimal 195 ) 196 if e := rows.Scan(&decisionContext, &groupLabel, &snapshotAt, &selected, &total, &selectionRate, &impactRatio, &modelVersion); e != nil { 197 return nil, e 198 } 199 k := key{decisionContext, groupLabel} 200 fs := byBucket[k] 201 if fs == nil { 202 fs = &FairnessSeries{DecisionContext: decisionContext, GroupLabel: groupLabel} 203 byBucket[k] = fs 204 order = append(order, k) 205 } 206 ratioStr := "n/a" 207 if impactRatio != nil { 208 ratioStr = impactRatio.StringFixed(4) 209 } 210 fs.Points = append(fs.Points, FairnessPoint{ 211 SnapshotAt: snapshotAt.UTC().Format(time.RFC3339), 212 SelectionRate: selectionRate.StringFixed(4), 213 ImpactRatio: ratioStr, 214 Selected: selected, 215 Total: total, 216 ModelVersion: modelVersion, 217 }) 218 } 219 if err := rows.Err(); err != nil { 220 return nil, err 221 } 222 223 out := make([]FairnessSeries, 0, len(order)) 224 for _, k := range order { 225 fs := byBucket[k] 226 evaluateAlert(fs) 227 out = append(out, *fs) 228 } 229 return out, nil 230} 231 232// evaluateAlert sets the drift alert on a series. It alerts when the latest impact ratio is below 0.80, 233// or when the latest ratio dropped by more than driftDelta from the prior snapshot's ratio (a fall 234// toward disparate impact, flagged even while still above the threshold). Points with an n/a ratio (the 235// reference group, or a cut with too few known groups) do not contribute a ratio to the comparison. 236func evaluateAlert(fs *FairnessSeries) { 237 var ratios []decimal.Decimal 238 for _, p := range fs.Points { 239 if p.ImpactRatio == "n/a" { 240 continue 241 } 242 ratios = append(ratios, decimalOrZero(p.ImpactRatio)) 243 } 244 if len(ratios) == 0 { 245 return 246 } 247 latest := ratios[len(ratios)-1] 248 if latest.LessThan(fourFifths) { 249 fs.Alert = true 250 fs.AlertReason = "latest impact ratio " + latest.StringFixed(4) + " is below the 0.80 four fifths threshold" 251 return 252 } 253 if len(ratios) >= 2 { 254 prior := ratios[len(ratios)-2] 255 drop := prior.Sub(latest) 256 if drop.GreaterThan(driftDelta) { 257 fs.Alert = true 258 fs.AlertReason = "impact ratio dropped " + drop.StringFixed(4) + " from " + prior.StringFixed(4) + " to " + latest.StringFixed(4) + ", more than the 0.05 drift limit" 259 } 260 } 261} 262