diff options
| -rw-r--r-- | cmds/export.go | 77 | ||||
| -rw-r--r-- | cmds/export_test.go | 67 | ||||
| -rw-r--r-- | cmds/negcap.go | 34 | ||||
| -rw-r--r-- | cmds/negcap_test.go | 25 |
4 files changed, 182 insertions, 21 deletions
diff --git a/cmds/export.go b/cmds/export.go index 102c94a..203a7ef 100644 --- a/cmds/export.go +++ b/cmds/export.go @@ -22,14 +22,15 @@ import ( ) type ExportCommand struct { - DumpRaw string - Labels string - State string - RunOut string - RecFeedID int64 - NegDays int - NegRatio int - NegSeed int64 + DumpRaw string + Labels string + State string + RunOut string + RecFeedID int64 + NegDays int + NegRatio int + NegSeed int64 + NegFeedCapPct int } func (c *ExportCommand) Name() string { return "export" } @@ -43,7 +44,9 @@ func (c *ExportCommand) Init(args []string) error { Export labels from Miniflux into labels.jsonl: starred entries become positives (incremental via --state cursor), plus a seeded random sample of recent read+unstarred entries as negatives at --neg-ratio per positive. -Each run also writes a JSON summary to --run-out. +No single feed may hold more than --neg-feed-cap-pct percent of the +negative target (floored at 50 rows), so a few firehoses can't own the +negative class. Each run also writes a JSON summary to --run-out. With --dump-raw, instead fetch ~25 starred and ~100 read unstarred entries and write the raw API responses to DIR for offline development fixtures. @@ -58,6 +61,7 @@ and write the raw API responses to DIR for offline development fixtures. fs.IntVar(&c.NegDays, "neg-days", 45, "Only sample negatives published within this many days") fs.IntVar(&c.NegRatio, "neg-ratio", 20, "Target negatives per positive") fs.Int64Var(&c.NegSeed, "neg-seed", 20240301, "Seed for negative sampling (same data + same seed = same sample)") + fs.IntVar(&c.NegFeedCapPct, "neg-feed-cap-pct", 5, "Per-feed negative ceiling as % of target (0 = uncapped; floor 50)") return fs.Parse(args) } @@ -202,11 +206,15 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { } posTotal, negTotal := 0, 0 + // negPerFeed is each feed's existing negative count: the cap is on the + // feed's total share of the pool, so history counts against it. + negPerFeed := make(map[int64]int) for _, row := range existing { if row.Label == core.LabelPos { posTotal++ } else { negTotal++ + negPerFeed[row.FeedID]++ } } posTotal += counts.appended @@ -218,7 +226,7 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { fmt.Fprintln(stdout, "note: rec-feed stars without a matching labeled original were skipped") } - nc, err := c.exportNegatives(client, seenIDs, seenURLs, posTotal, negTotal, now, stdout) + nc, err := c.exportNegatives(client, seenIDs, seenURLs, negPerFeed, posTotal, negTotal, now, stdout) if err != nil { return err } @@ -226,14 +234,15 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { RanAt: now, DurationMs: time.Since(started).Milliseconds(), PositivesSeen: counts.seen, PositivesNew: counts.appended, PosTotal: posTotal, NegCandidates: nc.candidates, NegNew: nc.appended, - NegTotal: negTotal + nc.appended, IDDup: counts.idDup, URLDup: counts.urlDup, + NegTotal: negTotal + nc.appended, NegCapped: nc.capped, + IDDup: counts.idDup, URLDup: counts.urlDup, RecHit: counts.recHit, RecUnresolved: counts.recUnresolved, }) } // negCounts is what the negatives sampler reports back. type negCounts struct { - candidates, appended int + candidates, appended, capped int } // exportNegatives samples read, unstarred entries published within the @@ -242,17 +251,24 @@ type negCounts struct { // (either polarity) is excluded; sampling is seeded so the same candidate // pool + seed reproduce the same sample. // +// negPerFeed carries each feed's existing negative count and is updated as +// rows are added: the --neg-feed-cap-pct ceiling applies to a feed's total +// pool share, so candidates from feeds already at their ceiling are skipped +// and the budget flows to other feeds. A capped-out pool can therefore +// finish short of the ratio target, which is reported, not hidden. +// // The window is enforced server-side via published_after (Miniflux >= 2.0.49, // the same floor the rest of export requires): only in-window entries are // fetched, so the deep archive is never walked and no client-side order // assumption is needed (Miniflux's default order is published_at asc). -func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int64]bool, seenURLs map[string]bool, posTotal, negTotal int, now time.Time, stdout io.Writer) (negCounts, error) { +func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int64]bool, seenURLs map[string]bool, negPerFeed map[int64]int, posTotal, negTotal int, now time.Time, stdout io.Writer) (negCounts, error) { var nc negCounts need := c.NegRatio*posTotal - negTotal if need <= 0 || posTotal == 0 { fmt.Fprintf(stdout, "negatives: skipped (need=%d pos=%d)\n", need, posTotal) return nc, nil } + cap := feedNegCap(c.NegFeedCapPct, c.NegRatio, posTotal) cutoff := now.Add(-time.Duration(c.NegDays) * 24 * time.Hour) unstarred := false @@ -285,17 +301,27 @@ func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int } nc.candidates = len(cands) + // Seeded permutation walk: same candidate pool + seed = same sample. + // The ceiling filters the walk rather than reordering it, so enabling + // or disabling it can't perturb the sampled order of accepted rows. rng := rand.New(rand.NewSource(c.NegSeed)) perm := rng.Perm(len(cands)) - take := min(need, len(cands)) - rows := make([]core.LabelRow, 0, take) - for i := range take { - e := cands[perm[i]] + rows := make([]core.LabelRow, 0, min(need, len(cands))) + for _, idx := range perm { + if len(rows) >= need { + break + } + e := cands[idx] + if cap > 0 && negPerFeed[e.Feed.ID] >= cap { + nc.capped++ + continue + } rows = append(rows, core.LabelRow{ EntryID: e.ID, URL: e.URL, Title: e.Title, FeedID: e.Feed.ID, FeedTitle: e.Feed.Title, Label: core.LabelNeg, CapturedAt: now, }) + negPerFeed[e.Feed.ID]++ seenIDs[e.ID] = true seenURLs[core.NormalizeURL(e.URL)] = true } @@ -304,15 +330,23 @@ func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int } nc.appended = len(rows) - fmt.Fprintf(stdout, "negatives: candidates=%d appended=%d (target=%d:%d)\n", - nc.candidates, nc.appended, c.NegRatio, posTotal) + fmt.Fprintf(stdout, "negatives: candidates=%d appended=%d capped=%d (target=%d:%d cap=%s)\n", + nc.candidates, nc.appended, nc.capped, c.NegRatio, posTotal, capLabel(cap)) if nc.appended < need { - fmt.Fprintf(stdout, "note: only %d usable candidates in the last %d days; negatives short by %d\n", - nc.candidates, c.NegDays, need-nc.appended) + fmt.Fprintf(stdout, "note: negatives short by %d (candidates=%d in the last %d days, capped-out=%d)\n", + need-nc.appended, nc.candidates, c.NegDays, nc.capped) } return nc, nil } +// capLabel renders a per-feed ceiling for the run line ("off" when uncapped). +func capLabel(cap int) string { + if cap <= 0 { + return "off" + } + return fmt.Sprintf("%d/feed", cap) +} + // runSummary is the per-run JSON summary written for visibility; kept to // the latest run (overwrite, not append). type runSummary struct { @@ -324,6 +358,7 @@ type runSummary struct { NegCandidates int `json:"neg_candidates"` NegNew int `json:"neg_new"` NegTotal int `json:"neg_total"` + NegCapped int `json:"neg_capped"` IDDup int `json:"id_dup"` URLDup int `json:"url_dup"` RecHit int `json:"rec_hit"` diff --git a/cmds/export_test.go b/cmds/export_test.go index 0190d39..25eadbb 100644 --- a/cmds/export_test.go +++ b/cmds/export_test.go @@ -115,10 +115,13 @@ func runExportAgainst(t *testing.T, p *pagedServer, dir string, extra ...string) t.Setenv("MINIFLUX_URL", srv.URL) t.Setenv("MINIFLUX_TOKEN", "test-token") + // Ratio/dedup semantics are tested uncapped; the per-feed ceiling has + // its own dedicated test (TestExportNegativesFeedCap). args := append([]string{ "--labels", filepath.Join(dir, "labels.jsonl"), "--state", filepath.Join(dir, "export_state.json"), "--run-out", filepath.Join(dir, "export_run.json"), + "--neg-feed-cap-pct", "0", }, extra...) cmd := &ExportCommand{} @@ -347,6 +350,70 @@ func TestExportNegatives(t *testing.T) { } } +// The per-feed ceiling: one feed's negatives are bounded by the cap, the +// budget then flows to other feeds, and the shortfall (vs. the ratio +// target) is reported rather than silently padded. +func TestExportNegativesFeedCap(t *testing.T) { + now := time.Now() + starred := makeEntries(10, 1000, true, 7) + for i := range starred { + starred[i].PublishedAt = now.Add(-24 * time.Hour) + } + // A big firehose pool and a small quiet feed's worth, both in-window. + big := makeEntries(500, 5000, false, 8) + small := makeEntries(30, 9000, false, 9) + for i := range big { + big[i].PublishedAt = now.Add(-10 * 24 * time.Hour) + } + for i := range small { + small[i].PublishedAt = now.Add(-10 * 24 * time.Hour) + } + p := &pagedServer{starred: starred, read: append(big, small...)} + dir := t.TempDir() + + // 10 positives × ratio 20 = target 200 negatives. Ceiling = 5% × 200 = + // 10 → floor 50 wins; cap is 50/feed. + out := runExportAgainst(t, p, dir, "--neg-days", "100000", "--neg-feed-cap-pct", "5") + + labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl")) + if err != nil { + t.Fatal(err) + } + perFeed := map[int64]int{} + negN := 0 + for _, row := range labels { + if row.Label == core.LabelNeg { + perFeed[row.FeedID]++ + negN++ + } + } + if perFeed[8] != 50 { + t.Errorf("firehose feed should be capped at 50, got %d", perFeed[8]) + } + if perFeed[9] != 30 { + t.Errorf("quiet feed should give all 30 of its candidates, got %d", perFeed[9]) + } + if negN != 80 { + t.Errorf("total negatives = %d, want 80 (50 capped + 30 quiet)", negN) + } + if !strings.Contains(out, "capped=") || !strings.Contains(out, "negatives short by 120") { + t.Errorf("shortfall from capping should be reported:\n%s", out) + } + body, err := os.ReadFile(filepath.Join(dir, "export_run.json")) + if err != nil { + t.Fatal(err) + } + var rs runSummary + if err := json.Unmarshal(body, &rs); err != nil { + t.Fatal(err) + } + // All 500 firehose candidates walk the permutation; 50 are accepted, + // the other 450 are skipped by its ceiling (the quiet feed never caps). + if rs.NegCapped != 450 { + t.Errorf("neg_capped = %d, want 450", rs.NegCapped) + } +} + func TestExportNegativesSeededDeterministic(t *testing.T) { dir1, dir2 := t.TempDir(), t.TempDir() runExportAgainst(t, negFixture(4, 120, 0), dir1) diff --git a/cmds/negcap.go b/cmds/negcap.go new file mode 100644 index 0000000..f531b14 --- /dev/null +++ b/cmds/negcap.go @@ -0,0 +1,34 @@ +// Per-feed negative ceiling, shared by `export` (enforced while sampling) +// and `compact-labels` (enforced against existing history). +// +// Why: negatives are sampled from the read+unstarred pool, whose size is +// dominated by high-volume feeds. Without a ceiling, a handful of firehoses +// own most of the negative class, and the model learns "wire headline" as +// its main axis instead of "headline samsc liked". The ceiling is on each +// feed's *total* negatives (existing history + this run's additions), not +// just the run's sample: a per-run cap would only slow domination, not +// bound it. +package cmds + +// negFeedCapFloor is the absolute minimum ceiling. It keeps a young, +// still-small negative pool from capping feeds at a handful of rows: +// early training needs volume more than it needs balance. +const negFeedCapFloor = 50 + +// feedNegCap returns the maximum negatives any single feed may hold in the +// pool: pct% of the target (negRatio × posTotal), floored at negFeedCapFloor +// and never above the target itself. pct <= 0 means uncapped (0). +func feedNegCap(pct, negRatio, posTotal int) int { + if pct <= 0 { + return 0 + } + target := negRatio * posTotal + cap := pct * target / 100 + if cap < negFeedCapFloor { + cap = negFeedCapFloor + } + if cap > target { + cap = target + } + return cap +} diff --git a/cmds/negcap_test.go b/cmds/negcap_test.go new file mode 100644 index 0000000..5076163 --- /dev/null +++ b/cmds/negcap_test.go @@ -0,0 +1,25 @@ +// The shared per-feed negative ceiling: uncapped at pct 0, floored so a +// young pool isn't strangled, and clamped to the target itself. +package cmds + +import "testing" + +func TestFeedNegCap(t *testing.T) { + for _, tc := range []struct { + name string + pct, ratio, pos, want int + }{ + {"uncapped at 0", 0, 20, 500, 0}, + {"normal: 5% of 10000-target", 5, 20, 500, 500}, + {"floor wins for young pools", 5, 20, 10, 50}, // target 200, 5% = 10 → floor 50 + {"clamped to the target", 90, 20, 3, 54}, // target 60, 90% = 54 + {"floor can exceed a tiny target", 5, 20, 1, 20}, // target 20 → clamp to 20 + } { + t.Run(tc.name, func(t *testing.T) { + if got := feedNegCap(tc.pct, tc.ratio, tc.pos); got != tc.want { + t.Errorf("feedNegCap(%d, %d, %d) = %d, want %d", + tc.pct, tc.ratio, tc.pos, got, tc.want) + } + }) + } +} |