mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-15 23:22:59 +08:00
feat(scheduler): per-group profit control for token account admission
Group pricing (rate multiplier, peak windows, per-user overrides) and account cost (accounts.rate_multiplier) already live side by side, but nothing stops the scheduler from handing a request to an account whose cost multiplier exceeds what the group's pricing can profitably serve. Add an opt-in per-group profit gate that filters scheduling candidates by a margin rule, while ordering, scoring, stickiness and breakers keep working unchanged among qualified accounts. Admission rule: an account qualifies iff U <= D * (1 - min_margin - safety_buffer) within a small relative epsilon, where U is accounts.rate_multiplier (0 is legal; missing/negative/NaN/Inf are conservatively rejected as invalid) and D is the requester's effective downstream multiplier (user-group override ?? group default, times the group peak factor) frozen at the request's pricing instant. - groups gain profit_control_enabled / profit_min_margin / profit_safety_buffer (migration 191); the durable auth-cache invalidation trigger additionally watches the profit and pricing columns (migration 192) so out-of-band group edits cannot leave stale auth snapshots; GetByKeyForAuth explicitly projects the new columns and the API-key auth snapshot version is bumped to force a refresh of pre-existing snapshots - request-level pricing instant: token entry points install pricingAt into ctx; the profit threshold D and the RecordUsage peak factor read the same instant, so one request never changes price mid-flight across waits/retries/failover (media and unwired paths keep the existing record-time semantics) - the gate covers token requests on openai, anthropic, gemini, grok and antigravity groups: OpenAI-family handlers via WithOpenAIRequestPricingContext (responses incl. WS bridge, chat completions, messages, embeddings, alpha search), the shared gateway via WithGatewayTokenRequestPricing (messages, chat completions, responses, gemini model actions); composite groups cannot enable it directly; image/video/models/usage/count_tokens stay ungated and an explicit image-generation intent suppresses the gate end to end - post-slot recheck: after a slot is acquired the account is re-read via SchedulerSnapshotService.GetAccount (scheduler cache, then DB; only when both fail the check fails open with WARN + metric); a vetoed account releases its slot and joins the request's exclusion set for reselection; sticky bindings are written only after the final check passes, and an over-threshold sticky account is skipped, not deleted, so it comes back once its rate recovers - sticky-session cache contract: GatewayCache.GetSessionAccountID now returns ErrStickySessionNotFound on a miss (mapped from redis.Nil in the repository implementation, mirroring ErrRefreshTokenNotFound) so the profit sticky path can distinguish "no binding yet" from a real read failure without importing the cache driver in service code - cross-group re-entry (composite parent -> member group) resolves the gate against the member group and clears a stale parent gate instead of letting a foreign threshold veto accounts - per-platform/group activity counters (installs, threshold vetoes, invalid-rate vetoes, refresh failures) for observability - admin UI: profit-control section on the five platforms' group forms with percent input, validation and platform-switch reset; group create/update/duplicate normalize and validate the config at a single choke point - cmd/profit-preview: offline what-if tool that replays the production admission semantics over an exported config/account/override/model dump, reports per-model admitted-account counts under the default and the worst-case (lowest user override) D, and surfaces probe-sync staleness as warnings without affecting admission Tests: service unit coverage for gate resolution/veto/threshold epsilon/pricing instant/suppress marker/scheduler filtering and post-slot recheck (incl. -race on the profit surface), unit-tagged handler slot-recheck and capability-mapping regressions, sqlmock and real-PostgreSQL integration regressions for the GetByKeyForAuth projection and the migration-192 trigger watch list, API contract update, and frontend specs for the five-platform form helpers.
This commit is contained in:
Brisbanehuang
@@ -0,0 +1,219 @@
|
||||
// profit-preview 读取生产只读导出的 JSON（分组利润配置、账号倍率与探测状态、
|
||||
// 用户覆盖倍率、主力模型清单），复用线上 U/D/阈值判定做五平台离线预演。
|
||||
//
|
||||
// 用法：
|
||||
//
|
||||
// go run ./cmd/profit-preview -input dump.json [-assume-enabled] [-json]
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
)
|
||||
|
||||
type inputGroup struct {
|
||||
ID int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Platform string `json:"platform"`
|
||||
RateMultiplier float64 `json:"rate_multiplier"`
|
||||
SubscriptionType string `json:"subscription_type"`
|
||||
ProfitControlEnabled bool `json:"profit_control_enabled"`
|
||||
ProfitMinMargin float64 `json:"profit_min_margin"`
|
||||
ProfitSafetyBuffer float64 `json:"profit_safety_buffer"`
|
||||
PeakRateEnabled bool `json:"peak_rate_enabled"`
|
||||
PeakStart string `json:"peak_start"`
|
||||
PeakEnd string `json:"peak_end"`
|
||||
PeakRateMultiplier float64 `json:"peak_rate_multiplier"`
|
||||
}
|
||||
|
||||
type inputAccount struct {
|
||||
ID int64 `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Platform string `json:"platform"`
|
||||
Type string `json:"type"`
|
||||
RateMultiplier *float64 `json:"rate_multiplier"`
|
||||
Extra map[string]any `json:"extra"`
|
||||
ModelMapping map[string]string `json:"model_mapping"`
|
||||
}
|
||||
|
||||
type inputEntry struct {
|
||||
Group inputGroup `json:"group"`
|
||||
Accounts []inputAccount `json:"accounts"`
|
||||
UserOverrides map[string]*float64 `json:"user_overrides"`
|
||||
Models []string `json:"models"`
|
||||
}
|
||||
|
||||
type inputDoc struct {
|
||||
Groups []inputEntry `json:"groups"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
inputPath := flag.String("input", "", "生产只读导出 JSON 路径")
|
||||
assumeEnabled := flag.Bool("assume-enabled", false, "把当前关闭的支持平台分组按保存配置视为已启用")
|
||||
jsonOut := flag.Bool("json", false, "以 JSON 输出完整报告（默认输出可读表格）")
|
||||
flag.Parse()
|
||||
if *inputPath == "" {
|
||||
fmt.Fprintln(os.Stderr, "usage: profit-preview -input dump.json [-assume-enabled] [-json]")
|
||||
os.Exit(2)
|
||||
}
|
||||
raw, err := os.ReadFile(*inputPath)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "read input: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
inputs, err := parsePreviewInputs(raw, *assumeEnabled)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "parse input: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
evalAt := time.Now()
|
||||
reports := service.PreviewProfitAdmission(inputs, evalAt)
|
||||
if len(reports) == 0 {
|
||||
fmt.Fprintln(os.Stderr, "input produced no preview reports")
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
if *jsonOut {
|
||||
enc := json.NewEncoder(os.Stdout)
|
||||
enc.SetIndent("", " ")
|
||||
if err := enc.Encode(map[string]any{"evaluated_at": evalAt, "reports": reports}); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "write output: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
fmt.Printf("利润门预演 @ %s（U=账号倍率；探测状态仅告警）\n", evalAt.Format(time.RFC3339))
|
||||
for _, report := range reports {
|
||||
fmt.Printf("\n== 分组 %d %s [%s] ==\n", report.GroupID, report.GroupName, report.Platform)
|
||||
fmt.Printf(" 利润门生效=%v 假定启用=%v | 默认 D=%.4f 阈值=%.4f | 最低有效 D=%.4f 阈值=%.4f\n",
|
||||
report.EffectiveGate, report.AssumedEnabled,
|
||||
report.DefaultD, report.ThresholdDefault, report.MinEffectiveD, report.ThresholdMinD)
|
||||
counts := map[string]int{}
|
||||
for _, v := range report.Verdicts {
|
||||
counts[v.Class]++
|
||||
rate := "-"
|
||||
if v.AccountRate != nil {
|
||||
rate = fmt.Sprintf("%.4f", *v.AccountRate)
|
||||
}
|
||||
flags := make([]string, 0, 2)
|
||||
if v.RejectedUnderMinD {
|
||||
flags = append(flags, "最低有效D下拒绝")
|
||||
}
|
||||
if len(v.Warnings) > 0 {
|
||||
flags = append(flags, strings.Join(v.Warnings, ","))
|
||||
}
|
||||
suffix := ""
|
||||
if len(flags) > 0 {
|
||||
suffix = " [" + strings.Join(flags, "; ") + "]"
|
||||
}
|
||||
fmt.Printf(" 账号 %-4d %-24s 平台=%-12s U=%-8s 来源=%-19s %s%s\n",
|
||||
v.AccountID, v.Name, v.Platform, rate, v.RateSource, v.Class, suffix)
|
||||
}
|
||||
fmt.Printf(" 分类合计: 准入=%d 利润不足=%d 倍率非法=%d\n",
|
||||
counts[service.ProfitPreviewClassAdmitted],
|
||||
counts[service.ProfitPreviewClassRejectedThreshold],
|
||||
counts[service.ProfitPreviewClassRejectedInvalidRate])
|
||||
models := make([]string, 0, len(report.RemainingByModel))
|
||||
for model := range report.RemainingByModel {
|
||||
models = append(models, model)
|
||||
}
|
||||
sort.Strings(models)
|
||||
for _, model := range models {
|
||||
fmt.Printf(" 模型 %-20s 利润门准入账号: 默认D=%d 最低有效D=%d\n",
|
||||
model, report.RemainingByModel[model], report.RemainingByModelMinD[model])
|
||||
}
|
||||
for _, model := range modelsWithZeroRemaining(report) {
|
||||
fmt.Printf(" 警告: 模型 %s 启用后利润门准入账号为 0\n", model)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func parsePreviewInputs(raw []byte, assumeEnabled bool) ([]service.ProfitPreviewGroupInput, error) {
|
||||
var doc inputDoc
|
||||
if err := json.Unmarshal(raw, &doc); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(doc.Groups) == 0 {
|
||||
return nil, fmt.Errorf("input contains no groups; check the export query and target configuration")
|
||||
}
|
||||
|
||||
inputs := make([]service.ProfitPreviewGroupInput, 0, len(doc.Groups))
|
||||
for i, entry := range doc.Groups {
|
||||
if entry.Group.ID <= 0 || strings.TrimSpace(entry.Group.Platform) == "" {
|
||||
return nil, fmt.Errorf("invalid group at index %d: id and platform are required", i)
|
||||
}
|
||||
group := &service.Group{
|
||||
ID: entry.Group.ID,
|
||||
Name: entry.Group.Name,
|
||||
Platform: entry.Group.Platform,
|
||||
Status: service.StatusActive,
|
||||
Hydrated: true,
|
||||
RateMultiplier: entry.Group.RateMultiplier,
|
||||
SubscriptionType: entry.Group.SubscriptionType,
|
||||
ProfitControlEnabled: entry.Group.ProfitControlEnabled,
|
||||
ProfitMinMargin: entry.Group.ProfitMinMargin,
|
||||
ProfitSafetyBuffer: entry.Group.ProfitSafetyBuffer,
|
||||
PeakRateEnabled: entry.Group.PeakRateEnabled,
|
||||
PeakStart: entry.Group.PeakStart,
|
||||
PeakEnd: entry.Group.PeakEnd,
|
||||
PeakRateMultiplier: entry.Group.PeakRateMultiplier,
|
||||
}
|
||||
accounts := make([]*service.Account, 0, len(entry.Accounts))
|
||||
for _, a := range entry.Accounts {
|
||||
account := &service.Account{
|
||||
ID: a.ID,
|
||||
Name: a.Name,
|
||||
Platform: a.Platform,
|
||||
Type: a.Type,
|
||||
RateMultiplier: a.RateMultiplier,
|
||||
Extra: a.Extra,
|
||||
}
|
||||
if len(a.ModelMapping) > 0 {
|
||||
mapping := make(map[string]any, len(a.ModelMapping))
|
||||
for k, v := range a.ModelMapping {
|
||||
mapping[k] = v
|
||||
}
|
||||
account.Credentials = map[string]any{"model_mapping": mapping}
|
||||
}
|
||||
accounts = append(accounts, account)
|
||||
}
|
||||
overrides := make(map[int64]float64, len(entry.UserOverrides))
|
||||
for userID, rate := range entry.UserOverrides {
|
||||
if rate == nil {
|
||||
continue
|
||||
}
|
||||
var id int64
|
||||
if _, err := fmt.Sscan(userID, &id); err == nil && id > 0 {
|
||||
overrides[id] = *rate
|
||||
}
|
||||
}
|
||||
inputs = append(inputs, service.ProfitPreviewGroupInput{
|
||||
Group: group,
|
||||
Accounts: accounts,
|
||||
UserOverrides: overrides,
|
||||
Models: entry.Models,
|
||||
AssumeEnabled: assumeEnabled,
|
||||
})
|
||||
}
|
||||
return inputs, nil
|
||||
}
|
||||
|
||||
func modelsWithZeroRemaining(report service.ProfitPreviewGroupReport) []string {
|
||||
var out []string
|
||||
for model, count := range report.RemainingByModel {
|
||||
if count == 0 {
|
||||
out = append(out, model)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestParsePreviewInputsIgnoresNullUserOverride(t *testing.T) {
|
||||
raw := []byte(`{
|
||||
"groups": [{
|
||||
"group": {
|
||||
"id": 50,
|
||||
"name": "preview",
|
||||
"platform": "openai",
|
||||
"rate_multiplier": 0.5,
|
||||
"subscription_type": "standard",
|
||||
"profit_control_enabled": false,
|
||||
"profit_min_margin": 0.1,
|
||||
"profit_safety_buffer": 0
|
||||
},
|
||||
"accounts": [{
|
||||
"id": 1,
|
||||
"name": "cheap",
|
||||
"platform": "openai",
|
||||
"type": "apikey",
|
||||
"rate_multiplier": 0.2
|
||||
}],
|
||||
"user_overrides": {"40": null, "41": 0.4},
|
||||
"models": ["gpt-test"]
|
||||
}]
|
||||
}`)
|
||||
|
||||
inputs, err := parsePreviewInputs(raw, true)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, inputs, 1)
|
||||
require.Equal(t, map[int64]float64{41: 0.4}, inputs[0].UserOverrides)
|
||||
require.True(t, inputs[0].AssumeEnabled)
|
||||
|
||||
report := service.PreviewProfitAdmission(inputs, time.Date(2026, 1, 15, 8, 30, 0, 0, time.UTC))[0]
|
||||
require.InDelta(t, 0.4, report.MinEffectiveD, 1e-12, "null 覆盖不能被解码成 0 倍率")
|
||||
require.InDelta(t, 0.36, report.ThresholdMinD, 1e-12)
|
||||
}
|
||||
|
||||
func TestParsePreviewInputsRejectsEmptyGroups(t *testing.T) {
|
||||
for _, raw := range [][]byte{
|
||||
[]byte(`{"groups":null}`),
|
||||
[]byte(`{"groups":[]}`),
|
||||
} {
|
||||
inputs, err := parsePreviewInputs(raw, false)
|
||||
require.ErrorContains(t, err, "input contains no groups")
|
||||
require.Nil(t, inputs)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user