feat(web): „Подобни договори" - ценови ориентир по CPV кохорта на страницата на договора

apps/web/app/routes/contract.tsx

@@ -1,7 +1,7 @@
 import { Link } from 'react-router';
 import { count, longDate, money, moneyBare, plural, signedMoney, signedPct } from '@sigma/shared';
 import { contractIdFromSlug, getContract } from '@sigma/db';
-import type { ContractDetail } from '@sigma/api-contract';
+import type { CohortBand, ContractDetail } from '@sigma/api-contract';
 import type { Route } from './+types/contract';
 import { Breadcrumbs } from '../components/Breadcrumbs';
 import { PageHeader } from '../components/PageHeader';
@@ -92,10 +92,24 @@ export async function loader({ params, context }: Route.LoaderArgs) {
   return { contract };
 }
 
+// Coarse cohort bands only (never a fake-precise "топ 4.7%") - @sigma/db cohortBand only claims a
+// fine band when the cohort is large enough and its percentiles are distinct (see cohort.ts).
+const COHORT_BAND_LABELS: Record<CohortBand, string> = {
+  top1: 'в най-горния 1% по стойност',
+  top5: 'в топ 5% по стойност',
+  top10: 'в топ 10% по стойност',
+  top25: 'в топ 25% по стойност',
+  'above-median': 'над медианата',
+  'at-median': 'около медианата',
+  'below-median': 'под медианата',
+  bottom25: 'сред най-ниските 25% по стойност',
+};
+
 const UNVERIFIED_VALUE_LABEL = 'стойност с непотвърдена достоверност';
 
 export default function Contract({ loaderData }: Route.ComponentProps) {
   const c = loaderData.contract;
+  const cohort = c.cohort;
   const v = c.value;
   const crumbId = c.unp || c.contractNumber || c.id;
   // Direct links to the day's raw ЦАИС ЕОП open-data files (storage.eop.bg) this record was
@@ -351,6 +365,34 @@ export default function Contract({ loaderData }: Route.ComponentProps) {
 
         <RiskIndicators contract={c} />
 
+        {cohort && (
+          <Section
+            id="similar"
+            title="Подобни договори"
+            hint="Стойността спрямо всички договори с чиста стойност в същия CPV сектор в базата."
+          >
+            <p>
+              <strong>{money(cohort.amountEur)}</strong> е{' '}
+              <strong>{COHORT_BAND_LABELS[cohort.band]}</strong> сред{' '}
+              {count(cohort.stats.pricedContracts)}{' '}
+              {plural(cohort.stats.pricedContracts, 'договор', 'договора')} в сектор „
+              {c.sector?.short ?? `CPV ${cohort.stats.division}`}“ (CPV {cohort.stats.division}).
+              Медианата за сектора е <strong>{money(cohort.stats.medianEur)}</strong>.
+            </p>
+            <p className="small muted">
+              Приблизителна позиция по предизчислени персентили на сектора, включващи и самия този
+              договор. Сравнението дава контекст на мащаба и не е оценка за нередност - голяма
+              поръчка може да е напълно обоснована. Използва различен метод от отчета за аномалии,
+              затова числата може леко да се разминават.
+            </p>
+            <p className="small muted">
+              <Link to={`/contracts?sector=${cohort.stats.division}&sort=value-desc`}>
+                Виж договорите в сектора →
+              </Link>
+            </p>
+          </Section>
+        )}
+
         <Section id="facts" title="Подробности">
           <FactsList
             rows={[

apps/web/app/routes/methodology.tsx

@@ -395,6 +395,19 @@ export default function Methodology({ loaderData }: Route.ComponentProps) {
                 Договори в чужда валута без намерен курс към датата на подписване се пазят като
                 записи, но се изключват от сумите в евро.
               </p>
+              <p>
+                <strong>„Подобни договори" на страницата на договор.</strong> Позицията („топ 5%",
+                „над медианата"…) сравнява стойността с всички договори с чиста стойност в същия
+                двуцифрен CPV сектор в базата (без времева граница - същият обхват като секторните
+                суми), по предизчислени персентили на сектора (включващи и самия договор). Показваме
+                сравнение само при поне 12 такива договора, а по-фина лента („топ 1%/5%/10%") — само
+                когато секторът е достатъчно голям и съответните персентили реално се различават, за
+                да не се появи фалшива точност при малка или изравнена по цена кохорта. Позицията е
+                приблизителна (широки стъпала, не точен процент), контекст за мащаба, а не оценка за
+                нередност, и ползва различен метод от отчета за аномалии, затова двете числа може
+                леко да се разминават. Договори с непотвърдена стойност не участват и не получават
+                сравнение.
+              </p>
             </section>
 
             <section className="section" aria-labelledby="identity">

docs/etl-pipeline-state.md

@@ -33,7 +33,7 @@ Scoped, идемпотентен дневен delta refresh. Заменя сам
 
 **Записва директно в live (served) таблици — няма blue-green swap:**
 - contracts (DELETE стари `c:e:`/`c:o:` от прозореца → INSERT нови с `amount_eur`)
-- `authority_totals`, `company_totals`, `flow_pairs`, `home_totals`, `sector_totals`, `facet_counts` (DELETE + REPLACE на засегнатите rollup-и)
+- `authority_totals`, `company_totals`, `flow_pairs`, `home_totals`, `sector_totals`, `facet_counts`, `cpv_division_stats` (DELETE + REPLACE на засегнатите rollup-и)
 - `search_index`, `data_freshness`
 
 **Извиква се от:**
@@ -101,7 +101,7 @@ Gate-ът е вързан **само в operator скриптовете** (`impo
 | Rollup | Поведение | Глобална консистентност |
 |--------|-----------|--------------------------|
 | `company_totals`, `authority_totals` | **scoped** към touched множеството (`refresh-slice.sql:1262`) | зависи от touched множеството |
-| `home_totals`, `sector_totals`, `facet_counts`, `flow_pairs`, `data_freshness` | **full-recompute** всеки run | по конструкция ✅ |
+| `home_totals`, `sector_totals`, `facet_counts`, `flow_pairs`, `cpv_division_stats`, `data_freshness` | **full-recompute** всеки run | по конструкция ✅ |
 
 Touched множеството се строи от **новата** атрибуция (`refresh-slice.sql:1198–1239`), след DELETE+INSERT на договорите. Contract id-то вгражда `bidder_key` (`refresh-slice.sql:527`), а DELETE-ът мачва само по `contract_number + tender` (ред 499).
 

packages/api-contract/src/index.ts

@@ -331,6 +331,10 @@ export interface ContractDetail {
   lots: ContractLots | null;
   /** Declared subcontractor from the АОП feed ("Подизпълнител"), sparse (~0.8% of contracts). */
   subcontractor: { name: string; eik: string | null; valueEur: number | null } | null;
+  /** „Подобни договори" value benchmark vs the contract's CPV-division cohort — null when no honest
+   *  comparison exists (suspect/absent value, no CPV, or a cohort below the minimum). Computed inside
+   *  getContract from the row it already read, so it costs one extra rollup read, not a second scan. */
+  cohort: ContractCohortBenchmark | null;
   /** Published amendments (annexes), oldest first — the recorded value history behind
    *  `value.signingEur` → `value.currentEur`. Empty when the contract has no annexes. */
   amendments: AmendmentEntry[];
@@ -341,6 +345,38 @@ export interface ContractRecord extends ContractDetail {
   sourceNames: { authority: string; bidder: string }; // verbatim source names
 }
 
+/** Precomputed value percentiles of one CPV division (cpv_division_stats rollup). */
+export interface CpvCohortStats {
+  division: string;
+  pricedContracts: number;
+  p25Eur: number;
+  medianEur: number;
+  p75Eur: number;
+  p90Eur: number;
+  p95Eur: number;
+  p99Eur: number;
+}
+
+/** Coarse position of one contract's value inside its CPV-division cohort. A „top X%" band is only
+ *  claimed when the cohort is large enough for that cut to be real AND the percentile anchors around
+ *  it are distinct (so a tie-collapsed or tiny cohort never yields a fake „top 1%"). */
+export type CohortBand =
+  | 'top1'
+  | 'top5'
+  | 'top10'
+  | 'top25'
+  | 'above-median'
+  | 'at-median'
+  | 'below-median'
+  | 'bottom25';
+
+/** The „Подобни договори" benchmark for the contract page - null when there is no honest cohort. */
+export interface ContractCohortBenchmark {
+  amountEur: number;
+  band: CohortBand;
+  stats: CpvCohortStats;
+}
+
 // ── Flows ───────────────────────────────────────────────────────────────────────────────────────
 
 export interface FlowPair {

packages/db/migrations/0004_cpv_division_stats.sql

@@ -0,0 +1,13 @@
+-- Per-CPV-division value percentiles for the contract-page "Подобни договори" benchmark.
+-- Filled by scripts/precompute.sql (full rebuild) and scripts/refresh-slice.sql (cron refresh);
+-- the contract page reads ONE row here instead of scanning its whole division per view.
+CREATE TABLE IF NOT EXISTS cpv_division_stats (
+  division TEXT PRIMARY KEY,
+  priced_contracts INTEGER NOT NULL,
+  p25_eur REAL NOT NULL,
+  median_eur REAL NOT NULL,
+  p75_eur REAL NOT NULL,
+  p90_eur REAL NOT NULL,
+  p95_eur REAL NOT NULL,
+  p99_eur REAL NOT NULL
+);

packages/db/src/precompute-cohort.test.ts

@@ -0,0 +1,132 @@
+/// <reference types="node" />
+import { execFileSync } from 'node:child_process';
+import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { dirname, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { describe, expect, it } from 'vitest';
+
+const root = resolve(dirname(fileURLToPath(import.meta.url)), '../../..');
+const migrations = [
+  resolve(root, 'packages/db/migrations/0000_init.sql'),
+  resolve(root, 'packages/db/migrations/0001_flow_pairs_bidder_index.sql'),
+  resolve(root, 'packages/db/migrations/0004_cpv_division_stats.sql'),
+];
+const precomputePath = resolve(root, 'scripts/precompute.sql');
+const refreshSlicePath = resolve(root, 'scripts/refresh-slice.sql');
+
+function sqlite(dbPath: string, sql: string): string {
+  return execFileSync('sqlite3', [dbPath], { input: sql, encoding: 'utf8' });
+}
+
+function sqliteJson<T>(dbPath: string, sql: string): T[] {
+  const out = execFileSync('sqlite3', ['-json', dbPath, sql], { encoding: 'utf8' }).trim();
+  return out ? (JSON.parse(out) as T[]) : [];
+}
+
+function readScript(dbPath: string, path: string): void {
+  execFileSync('sqlite3', ['-bail', dbPath], { input: `.read ${path}\n`, stdio: 'pipe' });
+}
+
+interface StatsRow {
+  division: string;
+  priced_contracts: number;
+  p25_eur: number;
+  median_eur: number;
+  p75_eur: number;
+  p90_eur: number;
+  p95_eur: number;
+  p99_eur: number;
+}
+
+describe('precompute cpv_division_stats', () => {
+  it('computes nearest-rank percentiles per division over the clean-value cohort', () => {
+    const dir = mkdtempSync(resolve(tmpdir(), 'sigma-cohort-'));
+    const dbPath = resolve(dir, 'test.sqlite');
+    try {
+      for (const migration of migrations) readScript(dbPath, migration);
+
+      // Division 45: 20 clean contracts valued 1000..20000 EUR, plus rows the cohort must exclude
+      // (suspect flag, NULL amount, zero amount, missing CPV). Division 30: a small 3-row cohort
+      // still gets a stats row - the MIN_COHORT floor is applied read-side, not here.
+      const stmts: string[] = [
+        `INSERT INTO authorities (id, name) VALUES ('auth:1', 'Възложител');`,
+        `INSERT INTO bidders (id, name) VALUES ('eik:1', 'Изпълнител');`,
+        `INSERT INTO tenders (id, source_id, title, authority_id, cpv_code, procedure_type)
+         VALUES ('t:45', 'UNP-45', 'Строителство', 'auth:1', '45000000', 'открита процедура'),
+                ('t:30', 'UNP-30', 'Техника', 'auth:1', '30200000', 'открита процедура'),
+                ('t:none', 'UNP-NONE', 'Без CPV', 'auth:1', NULL, 'открита процедура');`,
+      ];
+      for (let i = 1; i <= 20; i += 1) {
+        stmts.push(
+          `INSERT INTO contracts (id, tender_id, bidder_id, amount, amount_eur, value_flag)
+           VALUES ('c:45-${i}', 't:45', 'eik:1', ${i * 1000}, ${i * 1000}, 'ok');`,
+        );
+      }
+      stmts.push(
+        `INSERT INTO contracts (id, tender_id, bidder_id, amount, amount_eur, value_flag)
+         VALUES ('c:45-suspect', 't:45', 'eik:1', 9e9, 9e9, 'value_suspect'),
+                ('c:45-null', 't:45', 'eik:1', 5, NULL, 'ok'),
+                ('c:45-zero', 't:45', 'eik:1', 0, 0, 'ok'),
+                ('c:no-cpv', 't:none', 'eik:1', 7000, 7000, 'ok'),
+                ('c:30-1', 't:30', 'eik:1', 100, 100, 'ok'),
+                ('c:30-2', 't:30', 'eik:1', 200, 200, 'ok'),
+                ('c:30-3', 't:30', 'eik:1', 300, 300, 'ok');`,
+      );
+      sqlite(dbPath, stmts.join('\n'));
+
+      readScript(dbPath, precomputePath);
+
+      const rows = sqliteJson<StatsRow>(
+        dbPath,
+        'SELECT * FROM cpv_division_stats ORDER BY division;',
+      );
+      expect(rows).toHaveLength(2);
+
+      // n=20 → nearest-rank k = ceil(q*20): p25→5th, p50→10th, p75→15th, p90→18th, p95→19th, p99→20th.
+      expect(rows[1]).toEqual({
+        division: '45',
+        priced_contracts: 20,
+        p25_eur: 5000,
+        median_eur: 10000,
+        p75_eur: 15000,
+        p90_eur: 18000,
+        p95_eur: 19000,
+        p99_eur: 20000,
+      });
+
+      // n=3 → k = ceil(q*3): p25→1st, p50→2nd, p75/p90/p95/p99→3rd.
+      expect(rows[0]).toEqual({
+        division: '30',
+        priced_contracts: 3,
+        p25_eur: 100,
+        median_eur: 200,
+        p75_eur: 300,
+        p90_eur: 300,
+        p95_eur: 300,
+        p99_eur: 300,
+      });
+
+      // Idempotent: a re-run reflects current rows, no duplicates or drift.
+      readScript(dbPath, precomputePath);
+      expect(
+        sqlite(dbPath, "SELECT COUNT(*) FROM cpv_division_stats WHERE division = '45';").trim(),
+      ).toBe('1');
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+
+  it('keeps the refresh-slice rebuild identical to the precompute one', () => {
+    // Both files carry a full-rebuild copy of the same statement (small global rollup, like
+    // sector_totals). Extract INSERT INTO cpv_division_stats ... ; from each and compare with
+    // whitespace normalised, so the two paths cannot silently drift apart.
+    const extract = (path: string): string => {
+      const sql = readFileSync(path, 'utf8');
+      const m = /INSERT INTO cpv_division_stats[\s\S]*?;/.exec(sql);
+      if (!m) throw new Error(`no cpv_division_stats insert in ${path}`);
+      return m[0].replace(/\s+/g, ' ').trim();
+    };
+    expect(extract(refreshSlicePath)).toBe(extract(precomputePath));
+  });
+});