feat: ground competitive insights in peer metrics

2026-08-22 18:45:29 +08:00
parent 1a8a61cfe7
commit 4f91726edb
10 changed files with 497 additions and 86 deletions
--- a/internal/analytics/service.go
+++ b/internal/analytics/service.go
@@ -4,10 +4,10 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"slices"
 	"sync"
 	"time"

+	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/peercomparison"
 	"github.com/jackc/pgx/v5/pgxpool"
 )

@@ -135,21 +135,17 @@ func (s *Service) Start(ctx context.Context, merchantID int64, period Period, te
 			"reason": "TERMINAL_SCOPE_UNSUPPORTED", "eligiblePeerCount": 0, "combinedPeerSessionCount": 0,
 			"requiredPeerCount": 10, "requiredPeerSessionCount": 1000}
 	} else {
-		peerCount, peerSessions, err := s.peerEligibility(ctx, merchantID, period)
+		peerResult, err := s.peerComparison(ctx, merchantID, period)
 		if err != nil {
 			return Run{}, err
 		}
-		if peerCount < 10 || peerSessions < 1000 {
+		if peerResult.State != peercomparison.StateReady {
 			run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "SUPPRESSED", "metadata": metadata,
-				"reason": "INSUFFICIENT_COHORT", "eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions,
+				"reason": peerResult.Reason, "eligiblePeerCount": peerResult.EligiblePeerCount, "combinedPeerSessionCount": peerResult.CombinedPeerSessions,
 				"requiredPeerCount": 10, "requiredPeerSessionCount": 1000}
 		} else {
-			peerMetrics, err := s.peerMetrics(ctx, merchantID, period)
-			if err != nil {
-				return Run{}, err
-			}
 			run.Sections["peerComparison"] = map[string]any{"sectionId": "peer_comparison", "state": "READY", "metadata": metadata,
-				"data": map[string]any{"eligiblePeerCount": peerCount, "combinedPeerSessionCount": peerSessions, "metrics": peerMetrics}}
+				"data": map[string]any{"eligiblePeerCount": peerResult.EligiblePeerCount, "combinedPeerSessionCount": peerResult.CombinedPeerSessions, "metrics": peerResult.Metrics}}
 		}
 	}
 	s.mu.Lock()
@@ -395,85 +391,33 @@ func statusTrendQuery(periodType string) (string, int) {
 	FROM buckets b LEFT JOIN scoped s USING(bucket_start) GROUP BY b.bucket_start ORDER BY b.bucket_start`, map[string]int{"WEEK": 7, "JALALI_MONTH": 31}[periodType]
 }

-func (s *Service) peerEligibility(ctx context.Context, merchantID int64, p Period) (int64, int64, error) {
-	var peers, sessions int64
-	err := s.pool.QueryRow(ctx, `SELECT count(distinct ps.merchant_id),count(*) FROM payment_sessions ps JOIN merchants m ON m.id=ps.merchant_id
-	 WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.merchant_id<>$1 AND ps.created_at>=$2 AND ps.created_at<$3`, merchantID, p.StartAt, p.EndAt).Scan(&peers, &sessions)
-	if err != nil {
-		return 0, 0, fmt.Errorf("calculating peer eligibility: %w", err)
-	}
-	return peers, sessions, nil
-}
-
-type peerRow struct {
-	merchantID int64
-	values     [8]*float64
-}
-
-func (s *Service) peerMetrics(ctx context.Context, merchantID int64, p Period) ([]map[string]any, error) {
+func (s *Service) peerComparison(ctx context.Context, merchantID int64, p Period) (peercomparison.Result, error) {
 	query := `WITH session_stats AS (SELECT ps.merchant_id,ps.id,ps.status,ps.amount_rial,ps.fee_amount_rial,count(pa.id) attempts
 	 FROM payment_sessions ps LEFT JOIN payment_attempts pa ON pa.session_id=ps.id JOIN merchants m ON m.id=ps.merchant_id
 	 WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.created_at>=$2 AND ps.created_at<$3 GROUP BY ps.id),
-	 aggregate AS (SELECT merchant_id,100.0*count(*) filter(where status in ('Paid','Verified'))/nullif(count(*),0) success,
+	 aggregate AS (SELECT merchant_id,count(*) session_count,100.0*count(*) filter(where status in ('Paid','Verified'))/nullif(count(*),0) success,
 	 100.0*count(*) filter(where status='Verified')/nullif(count(*) filter(where status in ('Paid','Verified')),0) verification,
 	 100.0*count(*) filter(where attempts>1 and status in ('Paid','Verified'))/nullif(count(*) filter(where attempts>1),0) recovery,
 	 100.0*sum(fee_amount_rial) filter(where status in ('Paid','Verified'))/nullif(sum(amount_rial) filter(where status in ('Paid','Verified')),0) fee FROM session_stats GROUP BY merchant_id),
 	 timing AS (SELECT ps.merchant_id,percentile_disc(.5) within group(order by pa.init_time_ms)::float8 ip50,percentile_disc(.95) within group(order by pa.init_time_ms)::float8 ip95,
 	 percentile_disc(.5) within group(order by pa.verify_time_ms)::float8 vp50,percentile_disc(.95) within group(order by pa.verify_time_ms)::float8 vp95
 	 FROM payment_sessions ps JOIN payment_attempts pa ON pa.session_id=ps.id JOIN merchants m ON m.id=ps.merchant_id WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.created_at>=$2 AND ps.created_at<$3 GROUP BY ps.merchant_id)
-	 SELECT a.merchant_id,a.success::float8,a.verification::float8,a.recovery::float8,t.ip50,t.ip95,t.vp50,t.vp95,a.fee::float8 FROM aggregate a JOIN timing t USING(merchant_id)`
+	 SELECT a.merchant_id,a.session_count,a.success::float8,a.verification::float8,a.recovery::float8,t.ip50,t.ip95,t.vp50,t.vp95,a.fee::float8 FROM aggregate a LEFT JOIN timing t USING(merchant_id)`
 	rows, err := s.pool.Query(ctx, query, merchantID, p.StartAt, p.EndAt)
 	if err != nil {
-		return nil, fmt.Errorf("calculating peer metrics: %w", err)
+		return peercomparison.Result{}, fmt.Errorf("calculating peer metrics: %w", err)
 	}
 	defer rows.Close()
-	all := []peerRow{}
+	all := []peercomparison.MerchantValues{}
 	for rows.Next() {
-		var row peerRow
-		if err := rows.Scan(&row.merchantID, &row.values[0], &row.values[1], &row.values[2], &row.values[3], &row.values[4], &row.values[5], &row.values[6], &row.values[7]); err != nil {
-			return nil, fmt.Errorf("scanning peer metrics: %w", err)
+		var row peercomparison.MerchantValues
+		if err := rows.Scan(&row.MerchantID, &row.SessionCount, &row.Values[0], &row.Values[1], &row.Values[2], &row.Values[3], &row.Values[4], &row.Values[5], &row.Values[6], &row.Values[7]); err != nil {
+			return peercomparison.Result{}, fmt.Errorf("scanning peer metrics: %w", err)
 		}
 		all = append(all, row)
 	}
 	if err := rows.Err(); err != nil {
-		return nil, fmt.Errorf("iterating peer metrics: %w", err)
+		return peercomparison.Result{}, fmt.Errorf("iterating peer metrics: %w", err)
 	}
-	ids := []string{"payment_success_rate", "verification_completion_rate", "retry_recovery_rate", "init_time_ms_p50", "init_time_ms_p95", "verify_time_ms_p50", "verify_time_ms_p95", "paid_fee_rate"}
-	metrics := make([]map[string]any, 0, 8)
-	for index, id := range ids {
-		active := 0.0
-		peers := []float64{}
-		for _, row := range all {
-			if row.values[index] == nil {
-				continue
-			}
-			if row.merchantID == merchantID {
-				active = *row.values[index]
-			} else {
-				peers = append(peers, *row.values[index])
-			}
-		}
-		if len(peers) == 0 {
-			peers = append(peers, 0)
-		}
-		slices.Sort(peers)
-		median := peers[len(peers)/2]
-		if len(peers)%2 == 0 {
-			median = (peers[len(peers)/2-1] + median) / 2
-		}
-		lowerIsBetter := index >= 3
-		better := 0
-		for _, value := range peers {
-			if (!lowerIsBetter && active >= value) || (lowerIsBetter && active <= value) {
-				better++
-			}
-		}
-		percentile := float64(better) * 100 / float64(len(peers))
-		direction := "HIGHER_IS_BETTER"
-		if lowerIsBetter {
-			direction = "LOWER_IS_BETTER"
-		}
-		metrics = append(metrics, map[string]any{"metricId": id, "cohortMedian": fmt.Sprintf("%.2f", median), "merchantPerformancePercentile": fmt.Sprintf("%.2f", percentile), "direction": direction})
-	}
-	return metrics, nil
+	return peercomparison.Calculate(merchantID, false, all), nil
 }
--- a/internal/insight/avalai.go
+++ b/internal/insight/avalai.go
@@ -4,10 +4,12 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"unicode"

 	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/aiprovider"
+	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/peercomparison"
 )

 type AvalAIGenerator struct{ client *aiprovider.Client }
@@ -61,7 +63,7 @@ func (g *AvalAIGenerator) Generate(ctx context.Context, module Module, section S
 	if err := json.Unmarshal(raw, &generated); err != nil {
 		return Copy{}, providerUsage, errors.New("avalai: invalid page insight output")
 	}
-	copy, err := mergeGenerated(fallback, generated, periodContext.AllowedLabels)
+	copy, err := mergeGenerated(module, fallback, generated, periodContext.AllowedLabels, evidence)
 	return copy, providerUsage, err
 }

@@ -70,10 +72,13 @@ func insightInstructions(module Module) string {
 	if module == ModuleTransactions {
 		return base + " خلاصه باید دقیقاً وضعیت همان بازه انتخاب‌شده در صفحه تراکنش‌ها را توضیح دهد. Paid را پرداخت‌شده و در انتظار تأیید و Verified را تأیید نهایی بنامید؛ آن‌ها را با هم یا با Failed یکی نکنید. روند را توصیف کنید، اما برای تغییرها علت نسازید."
 	}
+	if module == ModulePeers {
+		return base + " برای جایگاه رقابتی، فقط از peerComparison استفاده کنید. وضعیت را میان کسب‌وکارهای مشابه و بر پایه داده‌های تجمیعی توضیح دهید، قوی‌ترین و ضعیف‌ترین دسته را مشخص کنید و برای ضعیف‌ترین دسته یک اقدام ساده پیشنهاد دهید. صدک‌ها و میانه‌ها را دقیقاً از شواهد کپی کنید و هیچ عدد تازه‌ای نسازید؛ از نام یا مقدار یک کسب‌وکار دیگر صحبت نکنید."
+	}
 	return base
 }

-func mergeGenerated(fallback Copy, generated generatedInsight, allowedPeriodLabels []string) (Copy, error) {
+func mergeGenerated(module Module, fallback Copy, generated generatedInsight, allowedPeriodLabels []string, evidence json.RawMessage) (Copy, error) {
 	known := map[string]struct{}{}
 	for _, id := range evidenceIDs(fallback) {
 		known[id] = struct{}{}
@@ -103,7 +108,11 @@ func mergeGenerated(fallback Copy, generated generatedInsight, allowedPeriodLabe
 		}
 	}
 	for _, text := range texts {
-		if strings.TrimSpace(text) == "" || len([]rune(text)) > 600 || containsUnexpectedDigit(text, allowedPeriodLabels) || !isPlainParagraph(text) {
+		unexpectedNumber := containsUnexpectedDigit(text, allowedPeriodLabels)
+		if module == ModulePeers {
+			unexpectedNumber = containsUnexpectedPeerNumber(text, allowedPeriodLabels, evidence)
+		}
+		if strings.TrimSpace(text) == "" || len([]rune(text)) > 600 || unexpectedNumber || !isPlainParagraph(text) {
 			return Copy{}, errors.New("avalai: unsafe page insight text")
 		}
 	}
@@ -132,6 +141,77 @@ func mergeGenerated(fallback Copy, generated generatedInsight, allowedPeriodLabe
 	return result, nil
 }

+func containsUnexpectedPeerNumber(value string, labels []string, evidence json.RawMessage) bool {
+	for _, label := range labels {
+		value = strings.ReplaceAll(value, label, "")
+	}
+	allowed := map[string]struct{}{}
+	var comparison peercomparison.Result
+	if json.Unmarshal(evidence, &struct {
+		Peer *peercomparison.Result `json:"peerComparison"`
+	}{Peer: &comparison}) == nil {
+		for _, metric := range comparison.Metrics {
+			for _, item := range []string{metric.CohortMedian, metric.MerchantPerformancePercentile} {
+				var number float64
+				if _, err := fmt.Sscanf(item, "%f", &number); err == nil {
+					allowed[fmt.Sprintf("%.2f", number)] = struct{}{}
+					allowed[fmt.Sprintf("%.0f", number)] = struct{}{}
+				}
+			}
+		}
+		allowed[fmt.Sprintf("%d", comparison.EligiblePeerCount)] = struct{}{}
+		allowed[fmt.Sprintf("%d", comparison.CombinedPeerSessions)] = struct{}{}
+	}
+	value = normalizeDigits(value)
+	for _, token := range numericTokens(value) {
+		if _, ok := allowed[token]; !ok {
+			return true
+		}
+	}
+	return false
+}
+
+func normalizeDigits(value string) string {
+	var builder strings.Builder
+	for _, r := range value {
+		switch {
+		case r >= '۰' && r <= '۹':
+			builder.WriteRune('0' + (r - '۰'))
+		case r >= '٠' && r <= '٩':
+			builder.WriteRune('0' + (r - '٠'))
+		case r == '٫':
+			builder.WriteByte('.')
+		case r == '٬':
+			// Persian thousands separators do not change the numeric token.
+		default:
+			builder.WriteRune(r)
+		}
+	}
+	return builder.String()
+}
+
+func numericTokens(value string) []string {
+	result := []string{}
+	start := -1
+	for index, r := range value {
+		isNumber := r >= '0' && r <= '9'
+		if r == '.' && start >= 0 {
+			continue
+		}
+		if isNumber && start < 0 {
+			start = index
+		}
+		if !isNumber && start >= 0 {
+			result = append(result, strings.TrimRight(value[start:index], "."))
+			start = -1
+		}
+	}
+	if start >= 0 {
+		result = append(result, strings.TrimRight(value[start:], "."))
+	}
+	return result
+}
+
 func hasExactPeriodReferences(generated generatedInsight, labels []string) bool {
 	if len(labels) < 2 {
 		return true
@@ -147,7 +227,7 @@ type generationPeriodContext struct {
 func periodGenerationContext(evidence json.RawMessage) generationPeriodContext {
 	var values struct {
 		ComparisonAvailable bool `json:"comparisonAvailable"`
-		SelectedPeriod struct {
+		SelectedPeriod      struct {
 			JalaliLabelFA string `json:"jalaliLabelFa"`
 		} `json:"selectedPeriod"`
 		PrecedingPeriod struct {
--- a/internal/insight/avalai_test.go
+++ b/internal/insight/avalai_test.go
@@ -1,8 +1,11 @@
 package insight

 import (
+	"encoding/json"
 	"strings"
 	"testing"
+
+	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/peercomparison"
 )

 func TestGeneratedInsightTextMustBeOnePlainParagraph(t *testing.T) {
@@ -55,3 +58,32 @@ func TestGeneratedInsightMustNameSelectedAndPrecedingPeriods(t *testing.T) {
 		t.Fatal("generic period wording was accepted")
 	}
 }
+
+func TestPeerInsightAllowsOnlyGroundedAggregateNumbers(t *testing.T) {
+	evidence, err := json.Marshal(map[string]any{
+		"peerComparison": peercomparison.Result{
+			State:                peercomparison.StateReady,
+			EligiblePeerCount:    10,
+			CombinedPeerSessions: 1000,
+			Metrics: []peercomparison.Metric{{
+				MetricID:                      "payment_success_rate",
+				CohortMedian:                  "72.50",
+				MerchantPerformancePercentile: "85.00",
+				Direction:                     "HIGHER_IS_BETTER",
+			}},
+		},
+	})
+	if err != nil {
+		t.Fatalf("marshal evidence: %v", err)
+	}
+	labels := []string{"مرداد ۱۴۰۵"}
+	if containsUnexpectedPeerNumber("در مرداد ۱۴۰۵، جایگاه شما صدک ۸۵٫۰۰ است.", labels, evidence) {
+		t.Fatal("grounded Persian percentile was rejected")
+	}
+	if containsUnexpectedPeerNumber("این مقایسه بر پایه ۱٬۰۰۰ نشست است.", labels, evidence) {
+		t.Fatal("grounded Persian session count was rejected")
+	}
+	if !containsUnexpectedPeerNumber("جایگاه شما صدک ۹۱ است.", labels, evidence) {
+		t.Fatal("invented percentile was accepted")
+	}
+}
--- a/internal/insight/builder.go
+++ b/internal/insight/builder.go
@@ -7,14 +7,16 @@ import (
 	"fmt"
 	"math"
 	"sort"
+	"strconv"
 	"time"

 	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/analytics"
+	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/peercomparison"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 )

-const AnalyticsVersion = "2.0.0"
+const AnalyticsVersion = "2.1.0"

 type Analysis struct {
 	Time       TimeContext
@@ -99,6 +101,14 @@ func (b *PostgreSQLBuilder) build(ctx context.Context, merchantID int64, termina
 	if err != nil {
 		return Analysis{}, err
 	}
+	var peerResult *peercomparison.Result
+	if module == ModulePeers {
+		result, peerErr := b.peerComparison(ctx, merchantID, terminalID, period, dataThrough)
+		if peerErr != nil {
+			return Analysis{}, peerErr
+		}
+		peerResult = &result
+	}
 	confidence := ConfidenceMedium
 	if timeContext.Freshness == FreshnessStale {
 		confidence = ConfidenceLow
@@ -108,7 +118,7 @@ func (b *PostgreSQLBuilder) build(ctx context.Context, merchantID int64, termina
 		confidence = ConfidenceInsufficientData
 	}
 	evidenceMap := map[string]any{
-		"schemaVersion": "1.0.0", "module": module, "comparisonAvailable": comparisonAvailable,
+		"schemaVersion": "2.0.0", "module": module, "comparisonAvailable": comparisonAvailable,
 		"time": timeContext, "privacy": map[string]any{"aggregateOnly": true, "peerSuppressionApplied": module == ModulePeers},
 	}
 	precedingPeriodLabel := ""
@@ -124,6 +134,9 @@ func (b *PostgreSQLBuilder) build(ctx context.Context, merchantID int64, termina
 		evidenceMap["latest30Days"] = latest
 		evidenceMap["previousYearEquivalent"] = previous
 	}
+	if peerResult != nil {
+		evidenceMap["peerComparison"] = peerResult
+	}
 	if period != nil && module == ModuleTransactions {
 		evidenceMap["statusDistribution"] = map[string]any{
 			"total": latest.Sessions, "paidAwaitingVerification": latest.Unverified, "verified": latest.Verified, "failed": latest.Failed, "reversed": latest.Reversed,
@@ -151,9 +164,9 @@ func (b *PostgreSQLBuilder) build(ctx context.Context, merchantID int64, termina
 	if err != nil {
 		return Analysis{}, fmt.Errorf("encoding insight evidence: %w", err)
 	}
-	copy := deterministicCopy(module, latest, previous, comparisonAvailable, timeContext, confidence)
+	copy := deterministicCopy(module, latest, previous, comparisonAvailable, timeContext, confidence, peerResult)
 	if period != nil {
-		copy = deterministicPeriodCopy(module, *period, precedingPeriodLabel, latest, previous, comparisonAvailable, timeContext, confidence)
+		copy = deterministicPeriodCopy(module, *period, precedingPeriodLabel, latest, previous, comparisonAvailable, timeContext, confidence, peerResult)
 	}
 	return Analysis{Time: timeContext, Confidence: confidence, Evidence: encoded, Copy: copy, Calendar: calendar}, nil
 }
@@ -188,6 +201,59 @@ func (b *PostgreSQLBuilder) aggregateRange(ctx context.Context, merchantID int64
 	return value, nil
 }

+func (b *PostgreSQLBuilder) peerComparison(ctx context.Context, merchantID int64, terminalID string, period *PeriodScope, dataThrough time.Time) (peercomparison.Result, error) {
+	if terminalID != "" {
+		return peercomparison.Calculate(merchantID, true, nil), nil
+	}
+	start, end := dataThrough.AddDate(0, 0, -29), dataThrough.AddDate(0, 0, 1)
+	if period != nil {
+		start, end = period.StartAt, period.EndAt
+	}
+	query := `WITH session_stats AS (
+		SELECT ps.merchant_id,ps.id,ps.status,ps.amount_rial,ps.fee_amount_rial,count(pa.id) attempts
+		FROM payment_sessions ps LEFT JOIN payment_attempts pa ON pa.session_id=ps.id
+		JOIN merchants m ON m.id=ps.merchant_id
+		WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.created_at >= $2 AND ps.created_at < $3
+		GROUP BY ps.id
+	), aggregate AS (
+		SELECT merchant_id,count(*) session_count,
+		100.0*count(*) FILTER(WHERE status IN ('Paid','Verified'))/nullif(count(*),0) success,
+		100.0*count(*) FILTER(WHERE status='Verified')/nullif(count(*) FILTER(WHERE status IN ('Paid','Verified')),0) verification,
+		100.0*count(*) FILTER(WHERE attempts>1 AND status IN ('Paid','Verified'))/nullif(count(*) FILTER(WHERE attempts>1),0) recovery,
+		100.0*sum(fee_amount_rial) FILTER(WHERE status IN ('Paid','Verified'))/nullif(sum(amount_rial) FILTER(WHERE status IN ('Paid','Verified')),0) fee
+		FROM session_stats GROUP BY merchant_id
+	), timing AS (
+		SELECT ps.merchant_id,
+		percentile_disc(.5) within group(order by pa.init_time_ms)::float8 ip50,
+		percentile_disc(.95) within group(order by pa.init_time_ms)::float8 ip95,
+		percentile_disc(.5) within group(order by pa.verify_time_ms)::float8 vp50,
+		percentile_disc(.95) within group(order by pa.verify_time_ms)::float8 vp95
+		FROM payment_sessions ps JOIN payment_attempts pa ON pa.session_id=ps.id JOIN merchants m ON m.id=ps.merchant_id
+		WHERE m.category_id=(SELECT category_id FROM merchants WHERE id=$1) AND ps.created_at >= $2 AND ps.created_at < $3
+		GROUP BY ps.merchant_id
+	)
+	SELECT a.merchant_id,a.session_count,a.success::float8,a.verification::float8,a.recovery::float8,
+		t.ip50,t.ip95,t.vp50,t.vp95,a.fee::float8
+	FROM aggregate a LEFT JOIN timing t USING(merchant_id)`
+	rows, err := b.pool.Query(ctx, query, merchantID, start, end)
+	if err != nil {
+		return peercomparison.Result{}, fmt.Errorf("loading peer insight evidence: %w", err)
+	}
+	defer rows.Close()
+	merchants := make([]peercomparison.MerchantValues, 0)
+	for rows.Next() {
+		var row peercomparison.MerchantValues
+		if err := rows.Scan(&row.MerchantID, &row.SessionCount, &row.Values[0], &row.Values[1], &row.Values[2], &row.Values[3], &row.Values[4], &row.Values[5], &row.Values[6], &row.Values[7]); err != nil {
+			return peercomparison.Result{}, fmt.Errorf("scanning peer insight evidence: %w", err)
+		}
+		merchants = append(merchants, row)
+	}
+	if err := rows.Err(); err != nil {
+		return peercomparison.Result{}, fmt.Errorf("iterating peer insight evidence: %w", err)
+	}
+	return peercomparison.Calculate(merchantID, false, merchants), nil
+}
+
 func ratio(numerator, denominator int64) any {
 	if denominator == 0 {
 		return nil
@@ -284,8 +350,11 @@ func horizon(module Module) int {
 	return 30
 }

-func deterministicCopy(module Module, latest, previous aggregate, comparison bool, tc TimeContext, confidence Confidence) Copy {
-	label := map[Module]string{ModuleOverview: "وضعیت کلی کسب‌وکار", ModulePayments: "عملکرد پرداخت", ModuleFees: "کارمزدها", ModulePeers: "مقایسه هم‌گروه", ModuleTransactions: "الگوی تراکنش‌ها", ModuleGrowth: "رشد"}[module]
+func deterministicCopy(module Module, latest, previous aggregate, comparison bool, tc TimeContext, confidence Confidence, peer ...*peercomparison.Result) Copy {
+	label := map[Module]string{ModuleOverview: "وضعیت کلی کسب‌وکار", ModulePayments: "عملکرد پرداخت", ModuleFees: "کارمزدها", ModulePeers: "جایگاه رقابتی", ModuleTransactions: "الگوی تراکنش‌ها", ModuleGrowth: "رشد"}[module]
+	if module == ModulePeers && len(peer) > 0 && peer[0] != nil {
+		return deterministicPeerCopy(peer[0], tc.DataThroughDate)
+	}
 	change := 0.0
 	if previous.Paid > 0 {
 		change = float64(latest.Paid-previous.Paid) / float64(previous.Paid) * 100
@@ -346,10 +415,13 @@ func deterministicTransactionCopy(period PeriodScope, precedingPeriodLabel strin
 	}
 }

-func deterministicPeriodCopy(module Module, period PeriodScope, precedingPeriodLabel string, latest, previous aggregate, comparison bool, tc TimeContext, confidence Confidence) Copy {
+func deterministicPeriodCopy(module Module, period PeriodScope, precedingPeriodLabel string, latest, previous aggregate, comparison bool, tc TimeContext, confidence Confidence, peer ...*peercomparison.Result) Copy {
 	if module == ModuleTransactions {
 		return deterministicTransactionCopy(period, precedingPeriodLabel, latest, previous, comparison, tc, confidence)
 	}
+	if module == ModulePeers && len(peer) > 0 && peer[0] != nil {
+		return deterministicPeerCopy(peer[0], period.JalaliLabelFA)
+	}

 	latestBody := fmt.Sprintf("در %s، %d تراکنش ثبت شد و %d پرداخت با موفقیت انجام شد.", period.JalaliLabelFA, latest.Sessions, latest.Paid)
 	switch module {
@@ -378,6 +450,56 @@ func deterministicPeriodCopy(module Module, period PeriodScope, precedingPeriodL
 	return copy
 }

+var peerMetricLabels = map[string]string{
+	"payment_success_rate": "موفقیت پرداخت", "verification_completion_rate": "تکمیل تأیید", "retry_recovery_rate": "بازیابی تلاش مجدد",
+	"init_time_ms_p50": "میانه زمان شروع", "init_time_ms_p95": "زمان شروع پرتکرار", "verify_time_ms_p50": "میانه زمان تأیید", "verify_time_ms_p95": "زمان تأیید پرتکرار", "paid_fee_rate": "نرخ کارمزد پرداخت موفق",
+}
+
+func deterministicPeerCopy(result *peercomparison.Result, periodLabel string) Copy {
+	if result.State != peercomparison.StateReady || len(result.Metrics) == 0 {
+		body := "برای مقایسه جایگاه رقابتی، هنوز دادهٔ تجمیعی کافی از کسب‌وکارهای مشابه وجود ندارد."
+		return Copy{Past: Narrative{Headline: "مقایسه با گروه مقایسه", Body: body, EvidenceID: []string{"peer-comparison"}}, LatestObserved: Narrative{Headline: "جایگاه رقابتی", Body: body, EvidenceID: []string{"peer-comparison"}}, Future7Days: Narrative{Headline: "برداشت کوتاه‌مدت", Body: body, EvidenceID: []string{"peer-comparison"}}, Future30Days: Narrative{Headline: "اقدام پیشنهادی", Body: "با تکمیل دادهٔ گروه مقایسه، این بخش جایگاه شما را دقیق‌تر توضیح می‌دهد.", EvidenceID: []string{"peer-comparison"}}, Findings: []Finding{{EvidenceID: "peer-comparison", Title: "دادهٔ مقایسه", Summary: body}}, Options: []Option{{Title: "بررسی دوباره", Description: "پس از تکمیل دادهٔ گروه مقایسه، این تحلیل را دوباره بررسی کنید.", EvidenceID: []string{"peer-comparison"}}}}
+	}
+	strongest, weakest := result.Metrics[0], result.Metrics[0]
+	total := 0.0
+	for _, metric := range result.Metrics {
+		value := parsePercent(metric.MerchantPerformancePercentile)
+		total += value
+		if value > parsePercent(strongest.MerchantPerformancePercentile) {
+			strongest = metric
+		}
+		if value < parsePercent(weakest.MerchantPerformancePercentile) {
+			weakest = metric
+		}
+	}
+	overall := total / float64(len(result.Metrics))
+	strongLabel, weakLabel := peerMetricLabels[strongest.MetricID], peerMetricLabels[weakest.MetricID]
+	if strongLabel == "" {
+		strongLabel = strongest.MetricID
+	}
+	if weakLabel == "" {
+		weakLabel = weakest.MetricID
+	}
+	periodPrefix := ""
+	if periodLabel != "" {
+		periodPrefix = fmt.Sprintf("در %s، ", periodLabel)
+	}
+	position := fmt.Sprintf("%sمیانگین جایگاه شما در هشت شاخص گروه مقایسه صدک %.2f است؛ صدک بالاتر یعنی عملکرد بهتر در این مقایسه.", periodPrefix, overall)
+	strongBody := fmt.Sprintf("قوی‌ترین دستهٔ شما «%s» با صدک %s است.", strongLabel, strongest.MerchantPerformancePercentile)
+	weakBody := fmt.Sprintf("ضعیف‌ترین دستهٔ شما «%s» با صدک %s است؛ بهبود همین دسته بیشترین فرصت رشد را دارد.", weakLabel, weakest.MerchantPerformancePercentile)
+	action := fmt.Sprintf("روند «%s» را در بازه‌های بعد بررسی کنید و برای بهبود آن یک اقدام مشخص در فرایند پرداخت تعریف کنید.", weakLabel)
+	latest := position + " " + strongBody + " " + weakBody
+	return Copy{Past: Narrative{Headline: "مقایسه با گروه مقایسه", Body: position, EvidenceID: []string{"peer-position"}}, LatestObserved: Narrative{Headline: "جایگاه رقابتی", Body: latest, EvidenceID: []string{"peer-position", "peer-strongest", "peer-weakest"}}, Future7Days: Narrative{Headline: "برداشت کوتاه‌مدت", Body: strongBody, EvidenceID: []string{"peer-strongest"}}, Future30Days: Narrative{Headline: "اقدام پیشنهادی", Body: action, EvidenceID: []string{"peer-weakest", "peer-action"}}, Findings: []Finding{{EvidenceID: "peer-position", Title: "جایگاه کلی", Summary: position, ScoreBPS: int(overall * 100), Confidence: ConfidenceMedium}, {EvidenceID: "peer-strongest", Title: "قوی‌ترین دسته", Summary: strongBody, ScoreBPS: int(parsePercent(strongest.MerchantPerformancePercentile) * 100), Confidence: ConfidenceMedium}, {EvidenceID: "peer-weakest", Title: "فرصت رشد", Summary: weakBody, ScoreBPS: int(parsePercent(weakest.MerchantPerformancePercentile) * 100), Confidence: ConfidenceMedium}}, Options: []Option{{Title: "تمرکز بر فرصت رشد", Description: action, EvidenceID: []string{"peer-weakest", "peer-action"}}}}
+}
+
+func parsePercent(value string) float64 {
+	result, err := strconv.ParseFloat(value, 64)
+	if err != nil {
+		return 0
+	}
+	return result
+}
+
 func score(impact int64, confidence Confidence) int {
 	base := int(math.Min(8000, math.Log10(float64(max(int64(1), impact))+1)*1200))
 	weight := map[Confidence]int{ConfidenceHigh: 10000, ConfidenceMedium: 7500, ConfidenceLow: 4500, ConfidenceInsufficientData: 2000}[confidence]
--- a/internal/insight/builder_transaction_test.go
+++ b/internal/insight/builder_transaction_test.go
@@ -3,6 +3,8 @@ package insight
 import (
 	"strings"
 	"testing"
+
+	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/peercomparison"
 )

 func TestDeterministicTransactionCopyDistinguishesPaymentStatesWithoutCausalClaims(t *testing.T) {
@@ -27,6 +29,22 @@ func TestDeterministicTransactionCopyDistinguishesPaymentStatesWithoutCausalClai
 	}
 }

+func TestDeterministicPeerCopyNamesStrongestWeakestAndAction(t *testing.T) {
+	result := peercomparison.Result{State: peercomparison.StateReady, Metrics: []peercomparison.Metric{
+		{MetricID: "payment_success_rate", MerchantPerformancePercentile: "85.00"},
+		{MetricID: "verification_completion_rate", MerchantPerformancePercentile: "40.00"},
+	}}
+	copy := deterministicPeerCopy(&result, "مرداد ۱۴۰۵")
+	for _, expected := range []string{"مرداد ۱۴۰۵", "موفقیت پرداخت", "صدک 85.00", "تکمیل تأیید", "صدک 40.00"} {
+		if !strings.Contains(copy.LatestObserved.Body, expected) {
+			t.Fatalf("peer summary missing %q: %s", expected, copy.LatestObserved.Body)
+		}
+	}
+	if !strings.Contains(copy.Options[0].Description, "تکمیل تأیید") {
+		t.Fatalf("peer action did not target weakest category: %+v", copy.Options)
+	}
+}
+
 func TestDeterministicTransactionCopyHandlesEmptyPeriod(t *testing.T) {
 	copy := deterministicTransactionCopy(
 		PeriodScope{JalaliLabelFA: "امروز"},
--- a/internal/insight/model.go
+++ b/internal/insight/model.go
@@ -10,7 +10,7 @@ type Module string

 type Section string

-const PageInsightVersion = "page_insight_v7"
+const PageInsightVersion = "page_insight_v8"

 const (
 	ModuleOverview     Module = "overview"
--- a/internal/insight/service.go
+++ b/internal/insight/service.go
@@ -151,7 +151,7 @@ func (s *Service) EnqueueAll(ctx context.Context, reason string) (int, error) {
 		return 0, err
 	}
 	timeContext := BuildTimeContext(s.now(), dataThrough)
-	modules := []Module{ModuleOverview, ModulePayments, ModuleFees, ModulePeers, ModuleGrowth}
+	modules := []Module{ModuleOverview, ModulePayments, ModuleFees, ModulePeers, ModuleTransactions, ModuleGrowth}
 	created := 0
 	for _, scope := range scopes {
 		for _, module := range modules {
--- a/internal/peercomparison/calculator.go
+++ b/internal/peercomparison/calculator.go
@@ -0,0 +1,139 @@
+// Package peercomparison contains the privacy-bounded, merchant-weighted
+// comparison calculation shared by analytics and page insights.
+package peercomparison
+
+import (
+	"fmt"
+	"sort"
+)
+
+const (
+	MinimumPeerCount    = 10
+	MinimumPeerSessions = 1000
+)
+
+type State string
+
+const (
+	StateReady      State = "READY"
+	StateSuppressed State = "SUPPRESSED"
+)
+
+type MetricDefinition struct {
+	ID            string
+	LowerIsBetter bool
+}
+
+var metricDefinitions = []MetricDefinition{
+	{ID: "payment_success_rate"},
+	{ID: "verification_completion_rate"},
+	{ID: "retry_recovery_rate"},
+	{ID: "init_time_ms_p50", LowerIsBetter: true},
+	{ID: "init_time_ms_p95", LowerIsBetter: true},
+	{ID: "verify_time_ms_p50", LowerIsBetter: true},
+	{ID: "verify_time_ms_p95", LowerIsBetter: true},
+	{ID: "paid_fee_rate", LowerIsBetter: true},
+}
+
+func MetricDefinitions() []MetricDefinition {
+	return append([]MetricDefinition(nil), metricDefinitions...)
+}
+
+// MerchantValues contains one aggregate value per merchant. It intentionally
+// carries no merchant data other than the internal id used to exclude the
+// active merchant and the session count used for suppression.
+type MerchantValues struct {
+	MerchantID   int64
+	SessionCount int64
+	Values       [8]*float64
+}
+
+type Metric struct {
+	MetricID                      string `json:"metricId"`
+	CohortMedian                  string `json:"cohortMedian"`
+	MerchantPerformancePercentile string `json:"merchantPerformancePercentile"`
+	Direction                     string `json:"direction"`
+}
+
+type Result struct {
+	State                State    `json:"state"`
+	Reason               string   `json:"reason,omitempty"`
+	EligiblePeerCount    int      `json:"eligiblePeerCount"`
+	CombinedPeerSessions int64    `json:"combinedPeerSessionCount"`
+	Metrics              []Metric `json:"metrics,omitempty"`
+}
+
+func Calculate(activeMerchantID int64, terminalScoped bool, merchants []MerchantValues) Result {
+	result := Result{State: StateSuppressed}
+	if terminalScoped {
+		result.Reason = "TERMINAL_SCOPE_UNSUPPORTED"
+		return result
+	}
+	var active *MerchantValues
+	peers := make([]MerchantValues, 0, len(merchants))
+	for index := range merchants {
+		merchant := merchants[index]
+		if merchant.MerchantID == activeMerchantID {
+			active = &merchants[index]
+			continue
+		}
+		peers = append(peers, merchant)
+		result.CombinedPeerSessions += merchant.SessionCount
+	}
+	result.EligiblePeerCount = len(peers)
+	if len(peers) < MinimumPeerCount || result.CombinedPeerSessions < MinimumPeerSessions {
+		result.Reason = "INSUFFICIENT_COHORT"
+		return result
+	}
+	if active == nil {
+		result.Reason = "INSUFFICIENT_COHORT"
+		return result
+	}
+
+	result.State = StateReady
+	result.Metrics = make([]Metric, 0, len(metricDefinitions))
+	for index, definition := range metricDefinitions {
+		activeValue := active.Values[index]
+		values := make([]float64, 0, len(peers))
+		for _, peer := range peers {
+			if peer.Values[index] != nil {
+				values = append(values, *peer.Values[index])
+			}
+		}
+		// Keep the public shape stable while making unavailable metric input
+		// explicit through the existing zero representation.
+		if len(values) == 0 || activeValue == nil {
+			result.Metrics = append(result.Metrics, Metric{MetricID: definition.ID, CohortMedian: "0.00", MerchantPerformancePercentile: "0.00", Direction: direction(definition)})
+			continue
+		}
+		sort.Float64s(values)
+		median := values[len(values)/2]
+		if len(values)%2 == 0 {
+			median = (values[len(values)/2-1] + median) / 2
+		}
+		below, equal := 0, 0
+		for _, value := range values {
+			if definition.LowerIsBetter {
+				if *activeValue < value {
+					below++
+				} else if *activeValue == value {
+					equal++
+				}
+			} else if *activeValue > value {
+				below++
+			} else if *activeValue == value {
+				equal++
+			}
+		}
+		percentileBPS := (2*below + equal) * 5000 / len(values)
+		result.Metrics = append(result.Metrics, Metric{MetricID: definition.ID, CohortMedian: fmt.Sprintf("%.2f", median), MerchantPerformancePercentile: fmt.Sprintf("%.2f", float64(percentileBPS)/100), Direction: direction(definition)})
+	}
+	return result
+}
+
+func direction(definition MetricDefinition) string {
+	if definition.LowerIsBetter {
+		return "LOWER_IS_BETTER"
+	}
+	return "HIGHER_IS_BETTER"
+}
--- a/internal/peercomparison/calculator_test.go
+++ b/internal/peercomparison/calculator_test.go
@@ -0,0 +1,76 @@
+package peercomparison
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func pointer(value float64) *float64 { return &value }
+
+func TestCalculateAppliesPrivacyThresholdsAndExcludesActiveMerchant(t *testing.T) {
+	merchants := make([]MerchantValues, 0, 11)
+	merchants = append(merchants, MerchantValues{MerchantID: 70, SessionCount: 900, Values: [8]*float64{pointer(0)}})
+	for i := int64(1); i <= 10; i++ {
+		merchants = append(merchants, MerchantValues{MerchantID: i, SessionCount: 100, Values: [8]*float64{pointer(float64(i))}})
+	}
+	result := Calculate(70, false, merchants)
+	if result.State != StateReady || result.EligiblePeerCount != 10 || result.CombinedPeerSessions != 1000 {
+		t.Fatalf("unexpected ready result: %+v", result)
+	}
+	if result.Metrics[0].MerchantPerformancePercentile != "0.00" {
+		t.Fatalf("active merchant must be ranked below all peers: %+v", result.Metrics[0])
+	}
+	suppressed := Calculate(70, true, merchants)
+	if suppressed.State != StateSuppressed || suppressed.Reason != "TERMINAL_SCOPE_UNSUPPORTED" {
+		t.Fatalf("unexpected terminal result: %+v", suppressed)
+	}
+	insufficient := Calculate(70, false, merchants[:10])
+	if insufficient.State != StateSuppressed || insufficient.Reason != "INSUFFICIENT_COHORT" {
+		t.Fatalf("unexpected insufficient result: %+v", insufficient)
+	}
+}
+
+func TestCalculateUsesDirectionAwareMidRankAndStableMetricOrder(t *testing.T) {
+	merchants := make([]MerchantValues, 0, 11)
+	merchants = append(merchants, MerchantValues{MerchantID: 70, SessionCount: 1, Values: [8]*float64{pointer(5), pointer(5), pointer(5), pointer(5), pointer(5), pointer(5), pointer(5), pointer(5)}})
+	for i := int64(1); i <= 10; i++ {
+		value := float64(i)
+		merchants = append(merchants, MerchantValues{MerchantID: i, SessionCount: 100, Values: [8]*float64{pointer(value), pointer(value), pointer(value), pointer(value), pointer(value), pointer(value), pointer(value), pointer(value)}})
+	}
+	result := Calculate(70, false, merchants)
+	if len(result.Metrics) != 8 || result.Metrics[0].MetricID != "payment_success_rate" || result.Metrics[7].MetricID != "paid_fee_rate" {
+		t.Fatalf("unexpected metric order: %+v", result.Metrics)
+	}
+	if result.Metrics[0].MerchantPerformancePercentile != "45.00" {
+		t.Fatalf("higher-is-better midrank = %s, want 45.00", result.Metrics[0].MerchantPerformancePercentile)
+	}
+	if result.Metrics[3].MerchantPerformancePercentile != "55.00" || result.Metrics[3].Direction != "LOWER_IS_BETTER" {
+		t.Fatalf("lower-is-better direction/percentile = %+v", result.Metrics[3])
+	}
+}
+
+func TestCalculateKeepsPeerIdentitiesOutOfSerializedResult(t *testing.T) {
+	merchants := []MerchantValues{{MerchantID: 987654321, Values: [8]*float64{pointer(5)}}}
+	for i := int64(1); i <= 10; i++ {
+		merchants = append(merchants, MerchantValues{
+			MerchantID:   1000 + i,
+			SessionCount: 100,
+			Values:       [8]*float64{pointer(5)},
+		})
+	}
+	result := Calculate(987654321, false, merchants)
+	encoded, err := json.Marshal(result)
+	if err != nil {
+		t.Fatalf("marshal result: %v", err)
+	}
+	text := string(encoded)
+	for _, forbidden := range []string{"merchantId", "987654321", "1001"} {
+		if strings.Contains(text, forbidden) {
+			t.Fatalf("serialized peer result exposed %q: %s", forbidden, text)
+		}
+	}
+	if result.Metrics[0].MerchantPerformancePercentile != "50.00" {
+		t.Fatalf("all-tie percentile = %s, want 50.00", result.Metrics[0].MerchantPerformancePercentile)
+	}
+}