-- 05. Implementación 1 (DuckDB SQL) de los screens de DEFINICIONES_PREVIAS.md §2 (sellado ee85d2f5…, 21-oct 19:08:55).
-- Entrada: tablas p (01-02), et (03), lab, excl. Salida: tabla sc e interno/sc_impl1.parquet (interno: lleva ids de lote).
CREATE OR REPLACE TABLE s0 AS
SELECT p.id_canonical, p.licitacion_id, p.organo_clave, p.anio, p.cpv, p.tipo, p.clase, p.tipo_contrato, p.num_ofertas,
  p.mn, p.mx, p.adj, p.pbl_lote, p.empresa_clave, p.fref, p.motivo0,
  CASE WHEN p.num_ofertas = 3 THEN '3' WHEN p.num_ofertas <= 5 THEN '4-5' WHEN p.num_ofertas <= 9 THEN '6-9' ELSE '10+' END tramo,
  CASE WHEN p.pbl_lote > 0 THEN CASE WHEN abs(p.mn - p.pbl_lote) <= 0.001*p.pbl_lote THEN 1 ELSE 0 END END pnc,
  (p.mn = p.mx) sin_disp
FROM p WHERE p.motivo0 = 'ok';
--;
CREATE OR REPLACE TABLE s1 AS
SELECT *, (NOT sin_disp AND coalesce(pnc,0) = 0) en_pc,
  (mx - mn) / mn rango, (mx - adj) / adj sep,
  CASE WHEN pbl_lote > 0 AND adj / pbl_lote <= 1.25 THEN 1 - adj / pbl_lote END baja
FROM s0;
--;
-- mediana de la baja por mercado × tramo, en PC, celdas con >= 10 bajas
CREATE OR REPLACE TABLE medb AS
SELECT cpv, tipo, anio, tramo, median(baja) med_baja, count(baja) nb FROM s1 WHERE en_pc AND baja IS NOT NULL GROUP BY ALL HAVING count(baja) >= 10;
--;
-- REP_IGUAL: licitaciones con >= 3 lotes adjudicados (b, O/S/S, 2018-2025), cuotas por ganador distinto
CREATE OR REPLACE TABLE rep AS
WITH l AS (SELECT licitacion_id, coalesce(empresa_clave, adjudicatario_nombre) g, coalesce(adj, 0) a FROM b
           WHERE anio BETWEEN 2018 AND 2025 AND tipo_contrato IN ('Obras','Servicios','Suministros')),
nl AS (SELECT licitacion_id FROM l GROUP BY 1 HAVING count(*) >= 3),
sh AS (SELECT licitacion_id, g, sum(a) a FROM l JOIN nl USING (licitacion_id) WHERE g IS NOT NULL GROUP BY ALL),
tot AS (SELECT licitacion_id, sum(a) t, count(*) k FROM sh GROUP BY 1 HAVING count(*) >= 2 AND sum(a) > 0),
s AS (SELECT sh.licitacion_id, sh.a / tot.t s, tot.k FROM sh JOIN tot USING (licitacion_id))
SELECT x.licitacion_id, 1 - sum(abs(x.s - y.s)) / (2 * any_value(x.k)) rep_igual
FROM s x JOIN s y USING (licitacion_id) GROUP BY 1;
--;
-- ESTAB: órgano × CPV, Jaccard de ganadores entre t y t-1 (b, O/S/S, sin personas físicas, 2017-2025), >= 3 lotes en los dos
CREATE OR REPLACE TABLE est AS
WITH l AS (SELECT organo_clave, cpv, anio, coalesce(empresa_clave, adjudicatario_nombre) g FROM b
           WHERE anio BETWEEN 2017 AND 2025 AND tipo_contrato IN ('Obras','Servicios','Suministros') AND NOT persona_fisica),
n AS (SELECT organo_clave, cpv, anio, count(*) nl FROM l GROUP BY ALL HAVING count(*) >= 3),
w AS (SELECT DISTINCT l.organo_clave, l.cpv, l.anio, l.g FROM l JOIN n USING (organo_clave, cpv, anio) WHERE l.g IS NOT NULL),
inter AS (SELECT a.organo_clave, a.cpv, a.anio, count(*) ni FROM w a JOIN w b2 ON b2.organo_clave = a.organo_clave AND b2.cpv = a.cpv AND b2.anio = a.anio - 1 AND b2.g = a.g GROUP BY ALL),
k AS (SELECT organo_clave, cpv, anio, count(*) nk FROM w GROUP BY ALL)
SELECT k1.organo_clave, k1.cpv, k1.anio, coalesce(i.ni, 0) * 1.0 / (k1.nk + k0.nk - coalesce(i.ni, 0)) estab
FROM k k1 JOIN k k0 ON k0.organo_clave = k1.organo_clave AND k0.cpv = k1.cpv AND k0.anio = k1.anio - 1
LEFT JOIN inter i ON i.organo_clave = k1.organo_clave AND i.cpv = k1.cpv AND i.anio = k1.anio;
--;
-- UTE_RIVALES
CREATE OR REPLACE TABLE ute AS
WITH so AS (SELECT DISTINCT ute_clave, socio_nif FROM idn_ute_socios WHERE NOT coalesce(socio_persona_fisica, FALSE) AND socio_nif IS NOT NULL),
u2 AS (SELECT ute_clave FROM so GROUP BY 1 HAVING count(*) >= 2),
solos AS (SELECT DISTINCT empresa_clave, cpv, tipo FROM b WHERE empresa_clave IS NOT NULL)
SELECT s1.id_canonical, CASE WHEN count(DISTINCT so.socio_nif) FILTER (WHERE solos.empresa_clave IS NOT NULL) >= 2 THEN 1 ELSE 0 END ute_rivales
FROM s1 JOIN u2 ON u2.ute_clave = s1.empresa_clave JOIN so ON so.ute_clave = u2.ute_clave
LEFT JOIN solos ON solos.empresa_clave = so.socio_nif AND solos.cpv = s1.cpv AND solos.tipo = s1.tipo
WHERE s1.en_pc GROUP BY 1;
--;
CREATE OR REPLACE TABLE s2 AS
SELECT s1.*, CASE WHEN m.med_baja IS NOT NULL AND s1.baja IS NOT NULL THEN s1.baja - m.med_baja END baja_cond,
  r.rep_igual, e.estab, CASE WHEN s1.en_pc THEN coalesce(t.ute_rivales, 0) END ute_rivales,
  count(*) FILTER (WHERE s1.en_pc) OVER (PARTITION BY s1.cpv, s1.tipo, s1.anio) n_merc
FROM s1 LEFT JOIN medb m USING (cpv, tipo, anio, tramo) LEFT JOIN rep r USING (licitacion_id)
LEFT JOIN est e ON e.organo_clave = s1.organo_clave AND e.cpv = s1.cpv AND e.anio = s1.anio
LEFT JOIN ute t USING (id_canonical);
--;
-- percentiles orientados dentro del mercado (PC, mercados con >= 20 lotes); rango medio de empates
CREATE OR REPLACE TABLE s3t AS
WITH q AS (SELECT *, (en_pc AND n_merc >= 20) en_c,
   -1*rango o_rango, -1*sep o_sep, -1*baja_cond o_baja, rep_igual o_rep, estab o_est FROM s2)
SELECT q.*,
  CASE WHEN en_c THEN ((rank() OVER (PARTITION BY en_c, cpv, tipo, anio ORDER BY o_rango) + count(*) OVER (PARTITION BY en_c, cpv, tipo, anio ORDER BY o_rango RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, cpv, tipo, anio) END p_rango,
  CASE WHEN en_c THEN ((rank() OVER (PARTITION BY en_c, cpv, tipo, anio ORDER BY o_sep) + count(*) OVER (PARTITION BY en_c, cpv, tipo, anio ORDER BY o_sep RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, cpv, tipo, anio) END p_sep,
  CASE WHEN en_c AND o_baja IS NOT NULL THEN ((rank() OVER (PARTITION BY en_c, o_baja IS NULL, cpv, tipo, anio ORDER BY o_baja) + count(*) OVER (PARTITION BY en_c, o_baja IS NULL, cpv, tipo, anio ORDER BY o_baja RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, o_baja IS NULL, cpv, tipo, anio) END p_baja,
  CASE WHEN en_c AND o_rep IS NOT NULL THEN ((rank() OVER (PARTITION BY en_c, o_rep IS NULL, cpv, tipo, anio ORDER BY o_rep) + count(*) OVER (PARTITION BY en_c, o_rep IS NULL, cpv, tipo, anio ORDER BY o_rep RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, o_rep IS NULL, cpv, tipo, anio) END p_rep,
  CASE WHEN en_c AND o_est IS NOT NULL THEN ((rank() OVER (PARTITION BY en_c, o_est IS NULL, cpv, tipo, anio ORDER BY o_est) + count(*) OVER (PARTITION BY en_c, o_est IS NULL, cpv, tipo, anio ORDER BY o_est RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, o_est IS NULL, cpv, tipo, anio) END p_est,
  -- sensibilidad (iii): percentiles dentro de mercado × tramo (celdas con >= 20)
  CASE WHEN en_c AND count(*) OVER (PARTITION BY en_c, cpv, tipo, anio, tramo) >= 20 THEN ((rank() OVER (PARTITION BY en_c, cpv, tipo, anio, tramo ORDER BY o_rango) + count(*) OVER (PARTITION BY en_c, cpv, tipo, anio, tramo ORDER BY o_rango RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, cpv, tipo, anio, tramo) END pt_rango,
  CASE WHEN en_c AND count(*) OVER (PARTITION BY en_c, cpv, tipo, anio, tramo) >= 20 THEN ((rank() OVER (PARTITION BY en_c, cpv, tipo, anio, tramo ORDER BY o_sep) + count(*) OVER (PARTITION BY en_c, cpv, tipo, anio, tramo ORDER BY o_sep RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, cpv, tipo, anio, tramo) END pt_sep,
  CASE WHEN en_c AND o_baja IS NOT NULL AND count(*) OVER (PARTITION BY en_c, o_baja IS NULL, cpv, tipo, anio, tramo) >= 20 THEN ((rank() OVER (PARTITION BY en_c, o_baja IS NULL, cpv, tipo, anio, tramo ORDER BY o_baja) + count(*) OVER (PARTITION BY en_c, o_baja IS NULL, cpv, tipo, anio, tramo ORDER BY o_baja RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
       / count(*) OVER (PARTITION BY en_c, o_baja IS NULL, cpv, tipo, anio, tramo) END pt_baja
FROM q;
--;
CREATE OR REPLACE TABLE s4 AS
SELECT *,
  CASE WHEN p_rango IS NOT NULL THEN (p_rango + p_sep + coalesce(p_baja, 0)) / (2 + CASE WHEN p_baja IS NULL THEN 0 ELSE 1 END) END s3,
  CASE WHEN p_rango IS NOT NULL THEN (p_rango + p_sep + coalesce(p_baja,0) + coalesce(p_rep,0) + coalesce(p_est,0))
       / (2 + (p_baja IS NOT NULL)::INT + (p_rep IS NOT NULL)::INT + (p_est IS NOT NULL)::INT) END s5,
  CASE WHEN pt_rango IS NOT NULL THEN (pt_rango + pt_sep + coalesce(pt_baja, 0)) / (2 + CASE WHEN pt_baja IS NULL THEN 0 ELSE 1 END) END s3_tramo
FROM s3t;
--;
-- percentil de S3 dentro del mercado (para D2: señalado si >= 0,80). S3 redondeado a 12 decimales antes de rangos (desvío técnico
-- declarado: empates numéricos de 2e-16 según el orden de la suma; detectado en el cotejo con impl2)
CREATE OR REPLACE TABLE sc AS
SELECT s4.*, CASE WHEN s4.s3 IS NOT NULL THEN ((rank() OVER (PARTITION BY s4.s3 IS NULL, s4.cpv, s4.tipo, s4.anio ORDER BY round(s4.s3, 12)) + count(*) OVER (PARTITION BY s4.s3 IS NULL, s4.cpv, s4.tipo, s4.anio ORDER BY round(s4.s3, 12) RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) / 2.0 - 0.5)
   / count(*) OVER (PARTITION BY s4.s3 IS NULL, s4.cpv, s4.tipo, s4.anio) END pct_s3,
  et.pos_alim, et.pos_mil, et.pos_carr, et.pos_cant, et.m_alim, et.m_mil, et.m_carr, et.m_cant, et.m_obras, et.en_cantabria, et.excluido,
  (SELECT count(*) FROM lab WHERE lab.id_canonical = s4.id_canonical AND lab.var = 'V2') > 0 es_v2
FROM s4 JOIN et USING (id_canonical);
--;
COPY (SELECT * EXCLUDE (o_rango, o_sep, o_baja, o_rep, o_est) FROM sc) TO '/c/interno/sc_impl1.parquet' (FORMAT PARQUET);
--;
-- @ comprobaciones: filas, PC, mercados, nulos (sin etiqueta)
SELECT count(*) p, count(*) FILTER (WHERE en_pc) pc, count(*) FILTER (WHERE en_c) pc_merc20, count(s3) con_s3, count(p_baja) con_baja,
  count(rep_igual) FILTER (WHERE en_c) con_rep, count(estab) FILTER (WHERE en_c) con_est, sum(ute_rivales) FILTER (WHERE en_c) ute_riv_1,
  count(*) FILTER (WHERE pnc IS NOT NULL) con_pnc, sum(pnc) pnc_1, count(*) FILTER (WHERE sin_disp) sin_disp, count(DISTINCT (cpv, tipo, anio)) FILTER (WHERE en_c) mercados
FROM sc;
--;
-- @ distribución de los screens en PC con mercado (sin etiqueta)
SELECT 'rango' s, approx_quantile(rango, [0.1,0.5,0.9]) q FROM sc WHERE en_c UNION ALL SELECT 'sep', approx_quantile(sep, [0.1,0.5,0.9]) FROM sc WHERE en_c
UNION ALL SELECT 'baja_cond', approx_quantile(baja_cond, [0.1,0.5,0.9]) FROM sc WHERE en_c UNION ALL SELECT 'rep_igual', approx_quantile(rep_igual, [0.1,0.5,0.9]) FROM sc WHERE en_c
UNION ALL SELECT 'estab', approx_quantile(estab, [0.1,0.5,0.9]) FROM sc WHERE en_c UNION ALL SELECT 's3', approx_quantile(s3, [0.1,0.5,0.9]) FROM sc WHERE en_c;
--;
-- @ percentiles en [0,1] y media 0,5 por mercado (control)
SELECT min(p_rango), max(p_rango), avg(p_rango), min(p_baja), max(p_baja), avg(p_baja), avg(pct_s3), count(*) FILTER (WHERE pct_s3 >= 0.8)*1.0/count(pct_s3) frac80 FROM sc WHERE en_c;