From 849077f89425aa22ee3a16b3b109b473959ace02 Mon Sep 17 00:00:00 2001 From: Ayrat Hudaygulov Date: Sun, 3 May 2026 01:53:22 +0100 Subject: [PATCH] Add ML repeat-count feature behind ML_REPEAT_COUNT_ENABLED flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Repeated spam ("campaigns") is the dominant failure mode of the 30-day-window SDCA model: 54% of top-50 campaign hits have a max dormant gap >30 days, so the bot forgets a campaign between bursts and re-learns it on each retrain. Adding text_repeat_count as an input feature lets the model condition on "how often have we seen this exact text" — a precision-improving prior that survives dormant-gap cycles even at the current training window. The feature is gated by a new bot_setting ML_REPEAT_COUNT_ENABLED (default off). When off, the trained pipeline does not concatenate repeatCountF and inference skips the DB lookup, so behavior is identical to today. When on, training adds the column and inference populates it via DB.GetTextRepeatCount, which uses the new partial index idx_event_msg_text_md5 from V35 to keep the lookup cheap. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/VahterBanBot/Bot.fs | 23 +++++++++-- src/VahterBanBot/DB.fs | 39 ++++++++++++++++++- src/VahterBanBot/ML.fs | 24 ++++++++++-- src/VahterBanBot/Program.fs | 1 + src/VahterBanBot/Types.fs | 5 +++ .../V35__event_msg_text_md5_index.sql | 13 +++++++ tests/VahterBanBot.Tests/ContainerTestBase.fs | 16 ++++++++ .../MLTrainingPipelineTests.fs | 6 +++ 8 files changed, 118 insertions(+), 9 deletions(-) create mode 100644 src/vahter-bot/migrations/V35__event_msg_text_md5_index.sql diff --git a/src/VahterBanBot/Bot.fs b/src/VahterBanBot/Bot.fs index ca90bcc..2562bfa 100644 --- a/src/VahterBanBot/Bot.fs +++ b/src/VahterBanBot/Bot.fs @@ -683,16 +683,30 @@ type BotService( // Private members — ML / LLM auto-verdict // ----------------------------------------------------------------------- + /// Repeat-count lookup gated by feature flag. Skipped when the flag is + /// off so we don't pay the DB round-trip on the hot inference path; the + /// trained pipeline doesn't reference repeatCountF in that case anyway. + /// Window matches MlTrainInterval so the inference distribution lines up + /// with the training-time text_repeat_counts CTE in DB.MlData. + member private _.GetMlRepeatCount(text: string) = task { + if not botConfig.Value.MlRepeatCountEnabled || isNull text then + return 0 + else + let since = utcNow() - botConfig.Value.MlTrainInterval + return! db.GetTextRepeatCount(text, since) + } + /// Fast text-only ML check used to short-circuit before Azure OCR. /// Returns Some score when ML alone is confident the message is spam /// (records the ML score so callers don't double-record). Returns None /// for null text, low ML score, warning band, or old-user immunity — /// in which case the caller should run Azure OCR for any cache misses /// and then call GetAutoVerdict for the full verdict. - member private _.PreOcrMlCheck(msg: TgMessage, usrMsgCount: int) = task { + member private this.PreOcrMlCheck(msg: TgMessage, usrMsgCount: int) = task { if isNull msg.Text then return None else - match ml.Predict(msg.Text, usrMsgCount, msg.Entities) with + let! repeatCount = this.GetMlRepeatCount(msg.Text) + match ml.Predict(msg.Text, usrMsgCount, msg.Entities, repeatCount) with | None -> return None | Some prediction -> // Old-user immunity: defer to GetAutoVerdict so the @@ -707,8 +721,9 @@ type BotService( } /// Runs ML prediction + optional LLM triage, returns verdict. - member private _.GetAutoVerdict(msg: TgMessage, usrMsgCount: int) = task { - match ml.Predict(msg.Text, usrMsgCount, msg.Entities) with + member private this.GetAutoVerdict(msg: TgMessage, usrMsgCount: int) = task { + let! repeatCount = this.GetMlRepeatCount(msg.Text) + match ml.Predict(msg.Text, usrMsgCount, msg.Entities, repeatCount) with | None -> return None | Some prediction -> do! db.RecordMlScoredMessage(msg.ChatId, msg.MessageId, float prediction.Score, prediction.Score >= botConfig.Value.MlSpamThreshold) diff --git a/src/VahterBanBot/DB.fs b/src/VahterBanBot/DB.fs index 01f4e7a..3cb5514 100644 --- a/src/VahterBanBot/DB.fs +++ b/src/VahterBanBot/DB.fs @@ -24,6 +24,7 @@ type SpamOrHamDb = spam: bool less_than_n_messages: bool custom_emoji_count: int + text_repeat_count: int created_at: DateTime } type DbService(connString: string, timeProvider: TimeProvider) = @@ -586,6 +587,7 @@ WITH final_messages AS ( (data->>'userId')::BIGINT AS user_id, data->>'text' AS text, data->'rawMessage'->'entities' AS entities, + msg_text_md5, created_at FROM event WHERE event_type IN ('MessageReceived', 'MessageEdited') @@ -599,6 +601,16 @@ user_msg_counts AS ( FROM final_messages GROUP BY user_id ), +text_repeat_counts AS ( + -- How many times the same text appears in the training window. + -- Repeated spam blasts (campaigns) get a high count; one-off chatter + -- gets 1. Symmetric with the GetTextRepeatCount inference query so + -- training and inference see the same feature distribution. + SELECT msg_text_md5, COUNT(*)::INT AS text_repeat_count + FROM final_messages + WHERE msg_text_md5 IS NOT NULL + GROUP BY msg_text_md5 +), verdicts AS ( -- All verdict-bearing events, unified across message and moderation streams SELECT @@ -630,11 +642,13 @@ SELECT m.text, COALESCE(u.less_than_n_messages, TRUE) AS less_than_n_messages, (SELECT COUNT(*) FROM jsonb_array_elements(m.entities) ent WHERE ent->>'type' = 'custom_emoji')::INT AS custom_emoji_count, + COALESCE(rc.text_repeat_count, 1) AS text_repeat_count, MAX(m.created_at) AS created_at FROM final_messages m LEFT JOIN last_verdict v ON v.chat_id = m.chat_id AND v.message_id = m.message_id LEFT JOIN user_msg_counts u ON u.user_id = m.user_id -GROUP BY m.text, v.is_spam, u.less_than_n_messages, m.entities +LEFT JOIN text_repeat_counts rc ON rc.msg_text_md5 = m.msg_text_md5 +GROUP BY m.text, v.is_spam, u.less_than_n_messages, m.entities, rc.text_repeat_count ORDER BY MAX(m.created_at); """ @@ -642,6 +656,29 @@ ORDER BY MAX(m.created_at); return Array.ofSeq data } + /// Counts how many MessageReceived events with the same text exist + /// in the time window [since, NOW]. Used at inference time to populate + /// the repeat-count ML feature symmetrically with how training computes + /// it from text_repeat_counts CTE in MlData. + /// Backed by idx_event_msg_text_md5 (V35). + member _.GetTextRepeatCount(text: string, since: DateTime) : Task = + task { + use conn = new NpgsqlConnection(connString) + + //language=postgresql + let sql = + """ +SELECT COUNT(*)::INT +FROM event +WHERE event_type = 'MessageReceived' + AND msg_text_md5 = md5(@text) + AND created_at >= @since; +""" + + let! count = conn.ExecuteScalarAsync(sql, {| text = text; since = since |}) + return count + } + /// Saves a trained ML model to the database (singleton row, upsert). member _.SaveTrainedModel(modelStream: Stream) : Task = task { diff --git a/src/VahterBanBot/ML.fs b/src/VahterBanBot/ML.fs index aaf2683..ae9f057 100644 --- a/src/VahterBanBot/ML.fs +++ b/src/VahterBanBot/ML.fs @@ -23,6 +23,11 @@ type SpamOrHam = spam: bool lessThanNMessagesF: single moreThanNEmojisF: single + /// log1p(text_repeat_count). Only consumed by the pipeline when + /// MlRepeatCountEnabled was true at training time; otherwise the + /// trained pipeline doesn't reference this column and the value + /// is ignored. + repeatCountF: single weight: single createdAt: DateTime } @@ -105,7 +110,10 @@ type MachineLearning( createdAt = x.created_at weight = w moreThanNEmojisF = if x.custom_emoji_count > botConf.Value.MlCustomEmojiThreshold then 1.0f else 0.0f - lessThanNMessagesF = if x.less_than_n_messages then 1.0f else 0.0f } + lessThanNMessagesF = if x.less_than_n_messages then 1.0f else 0.0f + // log1p compresses the long tail (max ~200 → ~5.3) so SDCA + // doesn't over-weight a single very-repeated campaign. + repeatCountF = single (Math.Log(1.0 + float x.text_repeat_count)) } ) |> fun x -> if botConf.Value.MlTrainRandomSortData then @@ -117,10 +125,17 @@ type MachineLearning( let trainingData = trainTestSplit.TrainSet let testData = trainTestSplit.TestSet + let featureColumns = + [| + "TextFeaturized" + "lessThanNMessagesF" + "moreThanNEmojisF" + if botConf.Value.MlRepeatCountEnabled then "repeatCountF" + |] let featurePipeline = mlContext.Transforms.Text .FeaturizeText(outputColumnName = "TextFeaturized", inputColumnName = "text") - .Append(mlContext.Transforms.Concatenate(outputColumnName = "Features", inputColumnNames = [|"TextFeaturized"; "lessThanNMessagesF"; "moreThanNEmojisF"|])) + .Append(mlContext.Transforms.Concatenate(outputColumnName = "Features", inputColumnNames = featureColumns)) let dataProcessPipeline = let options = SdcaLogisticRegressionBinaryTrainer.Options( @@ -179,7 +194,7 @@ type MachineLearning( // if ML is ready (either disabled or model is trained) member _.IsReady = not botConf.Value.MlEnabled || predictionEngine.IsSome - member _.Predict(text: string, userMsgCount: int, entities: MessageEntity array) = + member _.Predict(text: string, userMsgCount: int, entities: MessageEntity array, repeatCount: int) = try match predictionEngine with | Some predictionEngine -> @@ -189,12 +204,13 @@ type MachineLearning( |> Option.defaultValue [||] |> Seq.filter (fun x -> x.Type = MessageEntityType.CustomEmoji) |> Seq.length - + predictionEngine.Predict { text = text spam = false lessThanNMessagesF = if userMsgCount < botConf.Value.MlTrainCriticalMsgCount then 1.0f else 0.0f moreThanNEmojisF = if emojiCount > botConf.Value.MlCustomEmojiThreshold then 1.0f else 0.0f + repeatCountF = single (Math.Log(1.0 + float repeatCount)) weight = 1.0f createdAt = timeProvider.GetUtcNow().UtcDateTime } |> Some diff --git a/src/VahterBanBot/Program.fs b/src/VahterBanBot/Program.fs index 2000882..56bdb17 100644 --- a/src/VahterBanBot/Program.fs +++ b/src/VahterBanBot/Program.fs @@ -108,6 +108,7 @@ let buildBotConf () = MlCustomEmojiThreshold = getSettingOr "ML_CUSTOM_EMOJI_THRESHOLD" "20" |> int MlStopWordsInChats = getSettingOr "ML_STOP_WORDS_IN_CHATS" "{}" |> fromJson MlWeightDecayK = getSettingOr "ML_WEIGHT_DECAY_K" "0" |> float + MlRepeatCountEnabled = getSettingOr "ML_REPEAT_COUNT_ENABLED" "false" |> bool.Parse MlOldUserMsgCount = getSettingOr "ML_OLD_USER_MSG_COUNT" "50" |> int // Reaction spam detection ReactionSpamEnabled = getSettingOr "REACTION_SPAM_ENABLED" "false" |> bool.Parse diff --git a/src/VahterBanBot/Types.fs b/src/VahterBanBot/Types.fs index b001059..45c15d6 100644 --- a/src/VahterBanBot/Types.fs +++ b/src/VahterBanBot/Types.fs @@ -252,6 +252,11 @@ type BotConfiguration = MlStopWordsInChats: Dictionary /// Time-decay weight parameter: w(t) = exp(-k * age_in_days). 0 = no decay (all weights 1.0). MlWeightDecayK: float + /// Feature flag: include text-repeat count as an ML feature. + /// Affects training only — inference always populates the field correctly, + /// so toggling without retrain leaves predictions consistent with the + /// currently loaded model's pipeline. + MlRepeatCountEnabled: bool /// Users with >= this many unique messages are immune from ML/LLM triage. MlOldUserMsgCount: int // Reaction spam detection diff --git a/src/vahter-bot/migrations/V35__event_msg_text_md5_index.sql b/src/vahter-bot/migrations/V35__event_msg_text_md5_index.sql new file mode 100644 index 0000000..cb2d78e --- /dev/null +++ b/src/vahter-bot/migrations/V35__event_msg_text_md5_index.sql @@ -0,0 +1,13 @@ +-- Partial index on event.msg_text_md5 for MessageReceived events. +-- Powers the per-message repeat-count lookup used by the ML inference path +-- (DbService.GetTextRepeatCount) and by the training-set SQL's +-- text_repeat_counts CTE. Without this index those become sequential scans +-- over the full event log (~700k+ rows). +-- +-- Partial WHERE matches the access pattern: we only ever count +-- MessageReceived rows with a non-null hash. msg_text_md5 is NULL for +-- events whose data has no text field (most non-MessageReceived events). + +CREATE INDEX IF NOT EXISTS idx_event_msg_text_md5 + ON event(msg_text_md5) + WHERE event_type = 'MessageReceived' AND msg_text_md5 IS NOT NULL; diff --git a/tests/VahterBanBot.Tests/ContainerTestBase.fs b/tests/VahterBanBot.Tests/ContainerTestBase.fs index 17b6f85..266680c 100644 --- a/tests/VahterBanBot.Tests/ContainerTestBase.fs +++ b/tests/VahterBanBot.Tests/ContainerTestBase.fs @@ -618,9 +618,25 @@ type MlEnabledVahterTestContainers() = /// Variant that DELIBERATELY skips fixture preload to exercise the production training pipeline /// end-to-end. Used by MLTrainingPipelineTests as a smoke test that training still produces a /// usable model (the most important property of the bot — autonomous spam detection). +/// +/// Also enables ML_REPEAT_COUNT_ENABLED=true so the smoke test exercises the +/// repeat-count feature path through training AND inference. The non-FF path +/// is the production default and is implicitly covered by every other ML test. type MlTrainingFromScratchTestContainers() = inherit VahterTestContainers(mlEnabled = true) + override this.SeedDatabase(connString: string) = + let baseSeed = base.SeedDatabase(connString) + task { + do! baseSeed + use conn = new NpgsqlConnection(connString) + do! conn.OpenAsync() + do! conn.ExecuteAsync( + "INSERT INTO bot_setting(key,value,type,feature_group) \ + VALUES('ML_REPEAT_COUNT_ENABLED','true','FEATURE_FLAG','ML')") + :> Task + } + override this.AfterStart() = task { // Same /ready wait, but never extract bytes — we don't want a throwaway fresh-train diff --git a/tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs b/tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs index a5f3dd5..f1166b1 100644 --- a/tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs +++ b/tests/VahterBanBot.Tests/MLTrainingPipelineTests.fs @@ -12,6 +12,12 @@ open BotTestInfra /// deliberately starts with no model in the DB — forcing the bot's prod /// MachineLearning.StartAsync to train end-to-end. /// +/// The fixture also enables ML_REPEAT_COUNT_ENABLED=true so this test +/// exercises the repeat-count feature path through training (conditional +/// Concatenate of repeatCountF) and inference (DB.GetTextRepeatCount lookup +/// gated by the FF in Bot.GetMlRepeatCount). Production default is FF=off, +/// covered implicitly by every other ML test (which uses the pinned model). +/// /// The training pipeline is the most important property of this bot (it's what /// kills 90% of spam autonomously), so we assert it can: /// 1. Train successfully from seed data within a reasonable timeout