1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
// Compact-labels command: one-time (but safely re-runnable) rebalance of
// labels.jsonl against the same per-feed negative ceiling `export` now
// enforces. Existing labels were written before the ceiling existed, so a
// handful of firehoses can already own a large share of the negative class;
// this trims them back.
//
// This is the one deliberate exception to labels.jsonl's append-only rule
// (see core/labels.go). It exists because the ceiling only bounds the
// future: without a compaction, pre-cap history keeps distorting training
// until it ages out — which it never does. Rules that keep it honest:
// positives are never touched; a feed's newest negatives are the ones kept
// (recency carries the signal); the original file is copied to a backup
// before anything is rewritten; and running it twice is a no-op.
package cmds
import (
"bytes"
"encoding/json"
"flag"
"fmt"
"io"
"os"
"sort"
"fluxrec/core"
)
type CompactLabelsCommand struct {
Labels string
NegRatio int
NegFeedCapPct int
Backup string
DryRun bool
}
func (c *CompactLabelsCommand) Name() string { return "compact-labels" }
func (c *CompactLabelsCommand) Init(args []string) error {
fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
fs.Usage = func() {
fmt.Fprint(fs.Output(), `Usage: fluxrec compact-labels [flags]
Trim per-feed negative over-representation in labels.jsonl down to the
same ceiling export enforces (--neg-feed-cap-pct of the target, floor 50).
Positives are untouched; each trimmed feed keeps its NEWEST negatives; the
original is copied to --backup first; re-running is a no-op. Use --dry-run
to see the effect without writing anything.
`)
fs.PrintDefaults()
}
fs.StringVar(&c.Labels, "labels", "labels.jsonl", "Path to labels.jsonl")
fs.IntVar(&c.NegRatio, "neg-ratio", 20, "Negatives-per-positive ratio the ceiling is computed against")
fs.IntVar(&c.NegFeedCapPct, "neg-feed-cap-pct", 5, "Per-feed negative ceiling as % of target (floor 50)")
fs.StringVar(&c.Backup, "backup", "", "Backup path for the original file (default: LABELS.bak)")
fs.BoolVar(&c.DryRun, "dry-run", false, "Report what would change without writing")
if err := fs.Parse(args); err != nil {
return err
}
if c.NegRatio < 1 {
return fmt.Errorf("--neg-ratio must be >= 1, got %d", c.NegRatio)
}
if c.NegFeedCapPct < 1 {
return fmt.Errorf("--neg-feed-cap-pct must be >= 1, got %d", c.NegFeedCapPct)
}
return nil
}
func (c *CompactLabelsCommand) Run(stdin io.Reader, stdout io.Writer) error {
rows, err := core.ReadLabels(c.Labels)
if err != nil {
return err
}
if len(rows) == 0 {
fmt.Fprintf(stdout, "compact-labels: %s is empty or missing; nothing to do\n", c.Labels)
return nil
}
posTotal := 0
for _, row := range rows {
if row.Label == core.LabelPos {
posTotal++
}
}
cap := feedNegCap(c.NegFeedCapPct, c.NegRatio, posTotal)
if cap <= 0 {
return fmt.Errorf("computed per-feed ceiling is 0 (positives=%d); check --neg-ratio/--neg-feed-cap-pct", posTotal)
}
// Per-feed negative row indexes, to decide what each overfull feed keeps.
negIdx := make(map[int64][]int)
for i, row := range rows {
if row.Label == core.LabelNeg {
negIdx[row.FeedID] = append(negIdx[row.FeedID], i)
}
}
drop := make(map[int]bool)
trimmed := 0
for _, idxs := range negIdx {
if len(idxs) <= cap {
continue
}
trimmed++
// Keep the newest `cap` rows for this feed; newest-first sort with a
// deterministic tiebreak, so repeated runs agree.
sorted := make([]int, len(idxs))
copy(sorted, idxs)
sort.Slice(sorted, func(a, b int) bool {
ra, rb := rows[sorted[a]], rows[sorted[b]]
if ra.CapturedAt.Equal(rb.CapturedAt) {
return ra.EntryID > rb.EntryID
}
return ra.CapturedAt.After(rb.CapturedAt)
})
for _, idx := range sorted[cap:] {
drop[idx] = true
}
}
kept := make([]core.LabelRow, 0, len(rows)-len(drop))
for i, row := range rows {
if !drop[i] {
kept = append(kept, row)
}
}
negTotal := len(rows) - posTotal
fmt.Fprintf(stdout, "compact-labels: rows %d → %d (positives %d kept; negatives %d → %d; feeds trimmed %d; cap %d/feed)\n",
len(rows), len(kept), posTotal, negTotal, negTotal-len(drop), trimmed, cap)
if len(drop) == 0 {
fmt.Fprintln(stdout, "nothing over the ceiling; labels.jsonl unchanged")
return nil
}
if c.DryRun {
fmt.Fprintf(stdout, "dry-run: would drop %d negatives (no files written)\n", len(drop))
return nil
}
backup := c.Backup
if backup == "" {
backup = c.Labels + ".bak"
}
orig, err := os.ReadFile(c.Labels)
if err != nil {
return fmt.Errorf("reading %s for backup: %w", c.Labels, err)
}
if err := os.WriteFile(backup, orig, 0o644); err != nil {
return fmt.Errorf("writing backup %s: %w", backup, err)
}
var buf bytes.Buffer
enc := json.NewEncoder(&buf)
for _, row := range kept {
if err := enc.Encode(row); err != nil {
return fmt.Errorf("encoding label row: %w", err)
}
}
if err := os.WriteFile(c.Labels, buf.Bytes(), 0o644); err != nil {
return fmt.Errorf("rewriting %s: %w", c.Labels, err)
}
fmt.Fprintf(stdout, "wrote %s (backup: %s); re-running export will sample fresh negatives under the ceiling\n",
c.Labels, backup)
return nil
}
|
