From c2dd299c2f2ebb1c4d5614087402d38173ed93ec Mon Sep 17 00:00:00 2001 From: Sam Scholten Date: Thu, 15 Oct 2026 08:00:39 +1000 Subject: improve time stamps --- cmds/export.go | 41 ++++++++++++++++++++++++++++------------ cmds/export_test.go | 46 +++++++++++++++++++++++++++++++++++++++++++-- cmds/relabel.go | 42 ++++++++++++++++++++++++++++++++++------- cmds/train.go | 32 +++++++++++++++++++++++++++++++ cmds/train_test.go | 54 +++++++++++++++++++++++++++++++++++++++++++++++++++++ justfile | 25 ++++++------------------- 6 files changed, 200 insertions(+), 40 deletions(-) diff --git a/cmds/export.go b/cmds/export.go index 9b153ee..994717d 100644 --- a/cmds/export.go +++ b/cmds/export.go @@ -7,8 +7,9 @@ // (see relabel.go), or — for a fluxrec pick whose original was never // exported — is recorded as its own positive. // Negatives: seeded random sample of recent read+unstarred entries, topped -// up to a fixed ratio of the positive pool. Also carries --dump-raw fixture -// capture. +// up to a fixed ratio of the positive pool. Rows are stamped with the entry's +// own changed_at (the read/star moment), not the run clock. Also carries +// --dump-raw fixture capture. package cmds import ( @@ -52,6 +53,10 @@ 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. +captured_at is each entry's own changed_at (when it was read or starred), +not the run clock, so a bulk mark-read exported days later keeps its real +position in the time-split validation window. + A star always wins over the read state that produced a negative: a starred entry already labeled negative is relabeled to positive in place (identity kept, fresh captured_at, original copied to LABELS.bak), and a star on a @@ -116,6 +121,20 @@ func saveExportState(path string, st exportState) error { return nil } +// captureTime is when a label was formed: the entry's own last-changed time — +// for a read+unstarred entry that is when samsc read it, for a starred one +// when he starred it — rather than the moment this export run happened to +// visit it. Export-time stamps made every run look like fresh evidence: a bulk +// mark-read on Monday, exported on Friday, landed in the newest slice and so +// dominated the time-split validation window, which is supposed to measure +// recent behaviour. Falls back to the run clock when the API reports none. +func captureTime(e miniflux.Entry, now time.Time) time.Time { + if e.ChangedAt.IsZero() { + return now + } + return e.ChangedAt.UTC() +} + // exportCounts is what the run reports to stdout. type exportCounts struct { seen, appended, idDup, urlDup, recHit, recAdded, relabeled int @@ -153,10 +172,9 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { var counts exportCounts var newRows []core.LabelRow - // Every star seen this run, so rows already in the store can be flipped - // to positive after the walk (relabelStars). - starIDs := make(map[int64]bool) - starURLs := make(map[string]bool) + // Every star seen this run, with its own timestamp, so rows already in + // the store can be flipped to positive after the walk (relabelStars). + var stars starsByKey maxChanged := state.ChangedAfter now := time.Now().UTC() @@ -173,8 +191,7 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { if ts := e.ChangedAt.Unix(); ts > maxChanged { maxChanged = ts } - starIDs[e.ID] = true - starURLs[core.NormalizeURL(e.URL)] = true + stars.add(e.ID, e.URL, captureTime(e, now)) // Stars on the recommendations feed are not original entries. // A star outranks whatever the original's read state said, so @@ -191,7 +208,7 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { newRows = append(newRows, core.LabelRow{ EntryID: e.ID, URL: e.URL, Title: e.Title, FeedID: e.Feed.ID, FeedTitle: e.Feed.Title, - Label: core.LabelPos, CapturedAt: now, + Label: core.LabelPos, CapturedAt: captureTime(e, now), }) seenIDs[e.ID] = true seenURLs[nu] = true @@ -212,7 +229,7 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { newRows = append(newRows, core.LabelRow{ EntryID: e.ID, URL: e.URL, Title: e.Title, FeedID: e.Feed.ID, FeedTitle: e.Feed.Title, - Label: core.LabelPos, CapturedAt: now, + Label: core.LabelPos, CapturedAt: captureTime(e, now), }) seenIDs[e.ID] = true seenURLs[nu] = true @@ -229,7 +246,7 @@ func (c *ExportCommand) runExport(stdout io.Writer) error { // as recent evidence. Rewritten in place, backup first — the rows the // dedup counters above skipped as duplicates are exactly the ones this // rescues. - if flipped := relabelStars(existing, starIDs, starURLs, now); flipped > 0 { + if flipped := relabelStars(existing, stars, now); flipped > 0 { if err := rewriteLabels(c.Labels, c.Labels+".bak", existing); err != nil { return err } @@ -365,7 +382,7 @@ func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int 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, + Label: core.LabelNeg, CapturedAt: captureTime(e, now), }) negPerFeed[e.Feed.ID]++ seenIDs[e.ID] = true diff --git a/cmds/export_test.go b/cmds/export_test.go index 0be1ac3..4483f10 100644 --- a/cmds/export_test.go +++ b/cmds/export_test.go @@ -224,6 +224,10 @@ func TestExportURLDedup(t *testing.T) { // no row at all. func TestExportRelabelsStarredNegative(t *testing.T) { p := &pagedServer{starred: makeEntries(1, 1000, true, 7)} + // The star happened two hours ago; the flip must carry that moment, not + // the export run's clock. + starChanged := time.Now().Add(-2 * time.Hour).UTC().Truncate(time.Second) + p.starred[0].ChangedAt = starChanged dir := t.TempDir() labelsPath := filepath.Join(dir, "labels.jsonl") @@ -256,8 +260,11 @@ func TestExportRelabelsStarredNegative(t *testing.T) { if got.EntryID != 1000 || got.Title != "Title 1000" || got.FeedID != 7 { t.Errorf("relabel must keep the row's identity, got %+v", got) } - if !got.CapturedAt.After(old) { - t.Errorf("captured_at = %v, want refreshed past %v", got.CapturedAt, old) + if !got.CapturedAt.Equal(starChanged) { + t.Errorf("captured_at = %v, want the star's own time %v", got.CapturedAt, starChanged) + } + if got.CapturedAt.Before(old) { + t.Errorf("captured_at = %v predates the negative's %v; the star never moved the row", got.CapturedAt, old) } // The pre-rewrite file is preserved as a backup. @@ -317,6 +324,41 @@ func TestExportRelabelsRecFeedStar(t *testing.T) { } } +// Labels carry each entry's own changed_at (when the judgment happened), not +// the export run's clock; an entry reporting no changed_at falls back to the +// run time. +func TestExportStampsChangedAt(t *testing.T) { + judgment := time.Now().Add(-9 * 24 * time.Hour).UTC().Truncate(time.Second) + starred := makeEntries(2, 1000, true, 7) + starred[0].ChangedAt = judgment + starred[1].ChangedAt = time.Time{} + p := &pagedServer{starred: starred} + dir := t.TempDir() + + started := time.Now() + runExportAgainst(t, p, dir) + finished := time.Now() + + labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl")) + if err != nil { + t.Fatal(err) + } + if len(labels) != 2 { + t.Fatalf("expected 2 rows, got %d", len(labels)) + } + byID := map[int64]core.LabelRow{} + for _, row := range labels { + byID[row.EntryID] = row + } + if !byID[1000].CapturedAt.Equal(judgment) { + t.Errorf("captured_at = %v, want the entry's changed_at %v", byID[1000].CapturedAt, judgment) + } + fallback := byID[1001].CapturedAt + if fallback.Before(started) || fallback.After(finished) { + t.Errorf("zero changed_at should fall back to the run clock, got %v (run %v→%v)", fallback, started, finished) + } +} + func TestExportRecFeedResolution(t *testing.T) { // Two starred entries ON the recommendations feed (feed id 99). rec := makeEntries(2, 3000, true, 99) diff --git a/cmds/relabel.go b/cmds/relabel.go index 9233daf..3411b0d 100644 --- a/cmds/relabel.go +++ b/cmds/relabel.go @@ -28,22 +28,50 @@ import ( "fluxrec/core" ) +// starsByKey is this run's stars, each with the moment the star happened. +// Keyed by entry ID and by normalized URL, because a star on a rec-feed item +// is only tied to the original row by its URL. +type starsByKey struct { + byID map[int64]time.Time + byURL map[string]time.Time +} + +func (s *starsByKey) add(id int64, url string, at time.Time) { + if s.byID == nil { + s.byID = make(map[int64]time.Time) + s.byURL = make(map[string]time.Time) + } + s.byID[id] = at + s.byURL[core.NormalizeURL(url)] = at +} + +// stamp returns when the row's entry was starred, if it was. +func (s starsByKey) stamp(row core.LabelRow) (time.Time, bool) { + if t, ok := s.byID[row.EntryID]; ok { + return t, true + } + if t, ok := s.byURL[core.NormalizeURL(row.URL)]; ok { + return t, true + } + return time.Time{}, false +} + // relabelStars flips negative rows to positives for entries starred since they -// were labeled. ids and urls are this run's stars: entry IDs as seen on the -// starred entries, and normalized URLs (which is how a star on a rec-feed item -// resolves to the original row). Rows are mutated in place; the count of -// flipped rows is returned. -func relabelStars(rows []core.LabelRow, ids map[int64]bool, urls map[string]bool, now time.Time) int { +// were labeled, stamping each with the star's own time (falling back to the +// run clock when a row's star time is unknown). Rows are mutated in place; the +// count of flipped rows is returned. +func relabelStars(rows []core.LabelRow, stars starsByKey, now time.Time) int { flipped := 0 for i := range rows { if rows[i].Label != core.LabelNeg { continue } - if !ids[rows[i].EntryID] && !urls[core.NormalizeURL(rows[i].URL)] { + at, ok := stars.stamp(rows[i]) + if !ok { continue } rows[i].Label = core.LabelPos - rows[i].CapturedAt = now + rows[i].CapturedAt = at flipped++ } return flipped diff --git a/cmds/train.go b/cmds/train.go index 234ef06..4e9a54f 100644 --- a/cmds/train.go +++ b/cmds/train.go @@ -122,6 +122,10 @@ func (c *TrainCommand) Run(stdin io.Reader, stdout io.Writer) error { fmt.Fprintf(stdout, "train: pos=%d neg=%d | val: pos=%d neg=%d | precision@15=%.3f youden_j=%.3f | vocab=%d → %s\n", report.TrainPos, report.TrainNeg, report.ValPos, report.ValNeg, report.PrecisionAt15, report.YoudensJ, report.VocabSize, c.ModelOut) + if report.ValStampShare >= 0.5 { + fmt.Fprintf(stdout, "note: %d val rows (%.0f%%) share captured_at %s; one batch shaped this validation — treat precision@15 as that batch's measurement, not the model's\n", + report.ValStampCount, report.ValStampShare*100, report.ValStamp.Format(time.RFC3339)) + } return nil } @@ -202,6 +206,11 @@ func (c *TrainCommand) trainAndEvaluate(trainRows, valRows []core.LabelRow, now } } + // Val concentration: how much of the validation window is one capture + // batch (a bulk action or a single flood-heavy export). High share means + // precision@15 measured that batch, not steady-state taste. + stampShare, stampCount, stampAt := valStampConcentration(valRows) + vecParams := map[string]any{ "ngram_min": vectorizer.NgramMin, "ngram_max": vectorizer.NgramMax, "min_df": vectorizer.MinDF, "max_df": vectorizer.MaxDF, "vocab_cap": vectorizer.VocabCap, @@ -242,12 +251,32 @@ func (c *TrainCommand) trainAndEvaluate(trainRows, valRows []core.LabelRow, now TrainWindowMin: minCap.UTC(), TrainWindowMax: maxCap.UTC(), ValFrac: c.ValFrac, VocabSize: len(vectorizer.OrderedVocab), PrecisionAt15: precisionAt15, YoudensJ: youdenJ, + ValStampShare: stampShare, ValStampCount: stampCount, ValStamp: stampAt.UTC(), PositiveScores: posStats, NegativeScores: negStats, VectorizerParams: vecParams, ModelParams: modelParams, } return model, report, nil } +// valStampConcentration returns the largest share of validation rows that +// share one captured_at, that stamp's row count, and the stamp itself. +// Deterministic: ties on count pick the newer stamp. +func valStampConcentration(valRows []core.LabelRow) (share float64, count int, stamp time.Time) { + if len(valRows) == 0 { + return 0, 0, time.Time{} + } + byStamp := make(map[time.Time]int) + for _, r := range valRows { + byStamp[r.CapturedAt]++ + } + for ts, n := range byStamp { + if n > count || (n == count && ts.After(stamp)) { + share, count, stamp = float64(n)/float64(len(valRows)), n, ts + } + } + return share, count, stamp +} + // downsampleBalance downsamples the majority class to a 1:1 ratio with the // minority, using a seeded RNG for reproducibility. If already balanced, // returns the inputs unchanged. @@ -423,6 +452,9 @@ type trainReport struct { VocabSize int `json:"vocab_size"` PrecisionAt15 float64 `json:"precision_at_15"` YoudensJ float64 `json:"youdens_j"` + ValStampShare float64 `json:"val_max_stamp_share"` + ValStampCount int `json:"val_max_stamp_count"` + ValStamp time.Time `json:"val_max_stamp"` PositiveScores map[string]float64 `json:"positive_scores"` NegativeScores map[string]float64 `json:"negative_scores"` VectorizerParams map[string]any `json:"vectorizer_params"` diff --git a/cmds/train_test.go b/cmds/train_test.go index 5a2ecee..2a3bbeb 100644 --- a/cmds/train_test.go +++ b/cmds/train_test.go @@ -221,6 +221,60 @@ func TestTrainModelConsumedByScore(t *testing.T) { } } +// Val concentration: when one capture batch dominates the validation window, +// the report must say so (share/count/stamp) and stdout must flag it. +func TestTrainReportsValStampConcentration(t *testing.T) { + dir := t.TempDir() + labelsPath := filepath.Join(dir, "labels.jsonl") + syntheticLabels(t, labelsPath, 50, 50) + + // Restamp the newest 25 rows to one captured_at: val-frac 0.3 puts + // 30 rows in val, 25 of them from this batch. + rows, err := core.ReadLabels(labelsPath) + if err != nil { + t.Fatal(err) + } + stamp := rows[len(rows)-1].CapturedAt + for i := len(rows) - 25; i < len(rows); i++ { + rows[i].CapturedAt = stamp + } + if err := rewriteLabels(labelsPath, "", rows); err != nil { + t.Fatal(err) + } + + cmd := &TrainCommand{} + if err := cmd.Init([]string{labelsPath, "--val-frac", "0.3", + "--model-out", filepath.Join(dir, "m.json"), + "--report-out", filepath.Join(dir, "r.json")}); err != nil { + t.Fatalf("Init: %v", err) + } + var out bytes.Buffer + if err := cmd.Run(nil, &out); err != nil { + t.Fatalf("Run: %v", err) + } + + body, err := os.ReadFile(filepath.Join(dir, "r.json")) + if err != nil { + t.Fatal(err) + } + var report trainReport + if err := json.Unmarshal(body, &report); err != nil { + t.Fatal(err) + } + if want := 25.0 / 30.0; report.ValStampShare != want { + t.Errorf("val_max_stamp_share = %v, want %v", report.ValStampShare, want) + } + if report.ValStampCount != 25 { + t.Errorf("val_max_stamp_count = %d, want 25", report.ValStampCount) + } + if !report.ValStamp.Equal(stamp) { + t.Errorf("val_max_stamp = %v, want %v", report.ValStamp, stamp) + } + if !strings.Contains(out.String(), "one batch shaped this validation") { + t.Errorf("stdout should flag the concentrated validation:\n%s", out.String()) + } +} + // Guard: too few validation positives must error loudly. func TestTrainGuardFewValPositives(t *testing.T) { dir := t.TempDir() diff --git a/justfile b/justfile index df3d9d5..03f8afe 100644 --- a/justfile +++ b/justfile @@ -15,12 +15,10 @@ vet: clean: rm -f fluxrec -# Quick smoke: train-free, needs a model.json (from `fluxrec train`, future) +# Quick smoke: train-free, needs a model.json score TITLE MODEL="model.json": echo "{{TITLE}}" | go run . score --model {{MODEL}} -# ── auth (same pattern as the mflux justfile) ── - # Print the two export lines to paste in your shell (just can't set them for you) auth: @echo 'export MFLUX_URL="https://your-miniflux-host" # base URL, no trailing slash' @@ -41,26 +39,16 @@ retrostar LIMIT="50": # Pruning audit: per-feed volume + star stats feeds DAYS="180" MIN_PER_DAY="0.05" MIN_SEEN="0": check-auth - go run . feeds --days 180 --min-per-day 0.05 --min-seen 0 - -# ── retrain loop ── -# Eyeball report.json (precision@15 >= 0.10) before `just ship SERVER` — -# retrain never deploys. + go run . feeds --days {{DAYS}} --min-per-day {{MIN_PER_DAY}} --min-seen {{MIN_SEEN}} -# Export new labels, train, print the report path +# Export, compact the negative pool to the per-feed ceiling, then train (never deploys) retrain: check-auth go run . export + go run . compact-labels go run . train labels.jsonl @echo "report: report.json — check precision@15 >= 0.10 before shipping" -# ── deploy lane (local PC → server) ── -# SERVER is an ssh config alias; the agent does auth. No hostnames in this -# file. Run `ship` only after eyeballing report.json — retrain never deploys. -# The server state dir is bind-mounted at /data (created once on the VPS; -# see README "Deployment"); fluxrec runs under the compose stack at -# ~/services on the server. - -# Ship a freshly trained model.json to the server state dir +# Ship a freshly trained model.json to SERVER's state dir (SERVER = ssh alias) deploy SERVER: scp model.json {{SERVER}}:~/fluxrec-data/ @@ -72,8 +60,7 @@ restart SERVER: ship SERVER: (deploy SERVER) (restart SERVER) @echo "model shipped; serve restarted — watch /api/status next run" -# Full model refresh: retrain + ship in one verb (no report gate — eyeball -# report.json first if unsure, or use `retrain` + `ship` separately) +# retrain + ship in one verb (no report gate; use retrain/ship separately if unsure) refresh SERVER: retrain (ship SERVER) @echo "refreshed: retrained locally + model shipped to {{SERVER}}" -- cgit v1.2.3