Bridge-and-Join-s/internal/lkgateway/npr.go

package lkgateway

// Модуль «НПР / УДС» — риск-монитор клиентов маржинальной торговли поверх базы
// экспорта QUIK (quik_export, схема npr). Перенесён «как есть» с 10.10.10.23
// (FastAPI ui/app.py) 14.09.2026; читает базу логином только для чтения.
// Включается настройкой NPR.Enabled и лицензионным признаком "npr".

import (
	"context"
	"embed"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"io/fs"
	"log"
	"math"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

//go:embed web/static/npr/*
var nprStaticFS embed.FS

// NPRSettings — модуль НПР/УДС.
type NPRSettings struct {
	Enabled bool   `json:"enabled"`
	DSN     string `json:"dsn"` // postgres://quik_ro:...@host:5432/quik_export
}

func (r *RuntimeConfig) UpdateNPR(s NPRSettings) error {
	r.mu.Lock()
	r.data.NPR = s
	r.data.UpdatedAt = time.Now().UTC()
	r.mu.Unlock()
	return r.save()
}

// nprAvailable — модуль включён в настройках и разрешён лицензией.
func nprAvailable(rc *RuntimeConfig) bool {
	if rc == nil {
		return false
	}
	s := rc.Snapshot().NPR
	return s.Enabled && strings.TrimSpace(s.DSN) != "" && moduleAllowed(rc, "npr")
}

var (
	nprPoolMu  sync.Mutex
	nprPool    *pgxpool.Pool
	nprPoolDSN string
)

func nprDB(rc *RuntimeConfig) (*pgxpool.Pool, error) {
	dsn := strings.TrimSpace(rc.Snapshot().NPR.DSN)
	if dsn == "" {
		return nil, fmt.Errorf("модуль НПР: не задано подключение к базе экспорта QUIK")
	}
	nprPoolMu.Lock()
	defer nprPoolMu.Unlock()
	if nprPool != nil && nprPoolDSN == dsn {
		return nprPool, nil
	}
	if nprPool != nil {
		nprPool.Close()
		nprPool = nil
	}
	cfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		return nil, err
	}
	cfg.MaxConns = 5
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	p, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		return nil, err
	}
	nprPool, nprPoolDSN = p, dsn
	return p, nil
}

var nprMSK = time.FixedZone("MSK", 3*3600)

var nprSort = map[string]string{
	"uds_asc":     "uds ASC NULLS LAST, client_code ASC",
	"uds_desc":    "uds DESC NULLS LAST, client_code ASC",
	"client_asc":  "client_code ASC",
	"client_desc": "client_code DESC",
	"ts_desc":     "ts DESC",
}

var nprTabs = map[string]string{
	// Маржинальный риск: клиенты с реально открытыми позициями.
	"margin": "COALESCE(init_margin,0) > 0 OR COALESCE(min_margin,0) > 0",
	// Дебиторская задолженность: позиций нет, но портфель в минусе.
	"debt": "COALESCE(init_margin,0) = 0 AND COALESCE(min_margin,0) = 0 AND COALESCE(portfolio_value,0) < 0",
}

type nprRow struct {
	ClientCode, MarginType, UDSFlag   string
	UDS, RCV1, RCV2, MarginReq        *float64
	PortfolioValue                    *float64
	InitMargin, MinMargin, CorrMargin *float64
	Status, RiskLevel                 *int
	TS                                time.Time
	Zone                              string
}

func (r nprRow) Class() string {
	switch {
	case r.UDSFlag == "CALL" || (r.UDS != nil && *r.UDS < 1):
		return "row-call"
	case r.UDS != nil && *r.UDS < 1.5:
		return "row-warn"
	case r.UDSFlag == "INF":
		return "row-inf"
	}
	return ""
}

func (r nprRow) Badge() (cls, text string) {
	switch {
	case r.UDSFlag == "CALL":
		return "badge-call", "МАРЖИН-КОЛЛ"
	case r.UDSFlag == "INF":
		return "badge-inf", "∞ избыточно"
	case r.UDS != nil && *r.UDS < 1:
		return "badge-call", "маржин-колл"
	case r.UDS != nil && *r.UDS < 1.5:
		return "badge-warn", "внимание"
	case r.UDS != nil:
		return "badge-ok", "норма"
	}
	return "", "—"
}
func (r nprRow) BadgeClass() string { c, _ := r.Badge(); return c }
func (r nprRow) BadgeText() string  { _, t := r.Badge(); return t }

func nprF(v *float64) string {
	if v == nil || math.IsNaN(*v) {
		return "—"
	}
	return strings.ReplaceAll(strconv.FormatFloat(*v, 'f', 2, 64), ".", ",")
}
func nprI(v *int) string {
	if v == nil {
		return ""
	}
	return strconv.Itoa(*v)
}

func nprParseF(s string) *float64 {
	s = strings.TrimSpace(strings.ReplaceAll(s, ",", "."))
	if s == "" {
		return nil
	}
	f, err := strconv.ParseFloat(s, 64)
	if err != nil {
		return nil
	}
	return &f
}

func nprWhere(q string, mt []string, umn, umx *float64, onlyCalls bool, tab string) (string, []any) {
	var cl []string
	var params []any
	cl = append(cl, "("+nprTabs[tab]+")")
	if q != "" {
		params = append(params, "%"+q+"%")
		cl = append(cl, fmt.Sprintf("client_code ILIKE $%d", len(params)))
	}
	if len(mt) > 0 {
		params = append(params, mt)
		cl = append(cl, fmt.Sprintf("margin_type = ANY($%d)", len(params)))
	}
	if umn != nil {
		params = append(params, *umn)
		cl = append(cl, fmt.Sprintf("uds >= $%d", len(params)))
	}
	if umx != nil {
		params = append(params, *umx)
		cl = append(cl, fmt.Sprintf("uds <= $%d", len(params)))
	}
	if onlyCalls {
		cl = append(cl, "uds IS NOT NULL AND uds < 1")
	}
	return strings.Join(cl, " AND "), params
}

func scanNPR(rows pgx.Rows) ([]nprRow, int, error) {
	var out []nprRow
	total := 0
	for rows.Next() {
		var r nprRow
		var mt, fl *string
		var zone *string
		if err := rows.Scan(&r.ClientCode, &mt, &r.UDS, &fl, &r.RCV1, &r.RCV2, &r.MarginReq, &r.PortfolioValue, &r.InitMargin, &r.MinMargin, &r.CorrMargin, &r.Status, &r.RiskLevel, &r.TS, &zone, &total); err != nil {
			return nil, 0, err
		}
		if mt != nil {
			r.MarginType = *mt
		}
		if fl != nil {
			r.UDSFlag = *fl
		}
		if zone != nil {
			r.Zone = *zone
		}
		out = append(out, r)
	}
	return out, total, rows.Err()
}

const nprCols = `client_code, margin_type, uds::float8, uds_flag, rcv1::float8, rcv2::float8, margin_requirement::float8, portfolio_value::float8, init_margin::float8, min_margin::float8, corrected_margin::float8, status, risk_level, ts, zone`

func nprParsePeriod(hours int, from, to string) (time.Time, time.Time, bool) {
	parse := func(s string) (time.Time, bool) {
		for _, f := range []string{"2006-01-02T15:04", "2006-01-02T15:04:05", "2006-01-02 15:04", "2006-01-02"} {
			if t, err := time.ParseInLocation(f, s, nprMSK); err == nil {
				return t, true
			}
		}
		return time.Time{}, false
	}
	now := time.Now()
	tf, okf := parse(from)
	tt, okt := parse(to)
	if okf || okt {
		if !okf {
			tf = now.Add(-2 * 365 * 24 * time.Hour)
		}
		if !okt {
			tt = now
		}
		return tf, tt, true
	}
	if hours <= 0 {
		hours = 24
	}
	return now.Add(-time.Duration(hours) * time.Hour), now, false
}

// registerNPR — маршруты модуля.
func registerNPR(mux *http.ServeMux, a *admin, rc *RuntimeConfig) {
	sub, _ := fs.Sub(nprStaticFS, "web/static/npr")
	mux.Handle("/admin/npr/static/", http.StripPrefix("/admin/npr/static/", http.FileServer(http.FS(sub))))

	guard := func(w http.ResponseWriter, r *http.Request) (*pgxpool.Pool, bool) {
		if !nprAvailable(rc) {
			http.Error(w, "модуль «НПР / УДС» не включён или не входит в лицензию", http.StatusForbidden)
			return nil, false
		}
		p, err := nprDB(rc)
		if err != nil {
			http.Error(w, "модуль НПР: "+err.Error(), http.StatusServiceUnavailable)
			return nil, false
		}
		return p, true
	}
	qp := func(r *http.Request) (q string, mt []string, umn, umx *float64, onlyCalls bool, sort, tab string, page int) {
		v := r.URL.Query()
		q = strings.TrimSpace(v.Get("q"))
		for _, m := range v["mt"] {
			if m = strings.TrimSpace(m); m != "" {
				mt = append(mt, m)
			}
		}
		umn, umx = nprParseF(v.Get("umn")), nprParseF(v.Get("umx"))
		onlyCalls = v.Get("only_calls") == "true" || v.Get("only_calls") == "on"
		sort = v.Get("sort")
		if _, ok := nprSort[sort]; !ok {
			sort = "uds_asc"
		}
		tab = v.Get("tab")
		if _, ok := nprTabs[tab]; !ok {
			tab = "margin"
		}
		page, _ = strconv.Atoi(v.Get("page"))
		if page < 1 {
			page = 1
		}
		return
	}

	mux.HandleFunc("/admin/npr", func(w http.ResponseWriter, r *http.Request) {
		if _, ok := guard(w, r); !ok {
			return
		}
		q, mt, _, _, onlyCalls, sort, tab, pg := qp(r)
		data := struct {
			page
			Q, Sort, Tab string
			MT           []string
			UMN, UMX     string
			OnlyCalls    bool
			Page         int
		}{nowPage("НПР / УДС", "npr"), q, sort, tab, mt, r.URL.Query().Get("umn"), r.URL.Query().Get("umx"), onlyCalls, pg}
		render(w, a.npr, data)
	})

	mux.HandleFunc("/admin/npr/rows", func(w http.ResponseWriter, r *http.Request) {
		p, ok := guard(w, r)
		if !ok {
			return
		}
		q, mt, umn, umx, onlyCalls, sort, tab, page := qp(r)
		limit := 30
		where, params := nprWhere(q, mt, umn, umx, onlyCalls, tab)
		params = append(params, limit, (page-1)*limit)
		sql := fmt.Sprintf(`WITH per_client AS (
			SELECT DISTINCT ON (client_code) %s FROM npr.v_current
			ORDER BY client_code, (COALESCE(abs(rcv1),0) + COALESCE(abs(rcv2),0)) DESC)
			SELECT %s, count(*) OVER () AS total FROM per_client WHERE %s ORDER BY %s LIMIT $%d OFFSET $%d`,
			nprCols, nprCols, where, nprSort[sort], len(params)-1, len(params))
		ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second)
		defer cancel()
		rows, err := p.Query(ctx, sql, params...)
		if err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		list, total, err := scanNPR(rows)
		rows.Close()
		if err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		pages := (total + limit - 1) / limit
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		w.Header().Set("Cache-Control", "no-store")
		_ = a.nprRows.ExecuteTemplate(w, "npr_rows", map[string]any{"Rows": list, "Total": total, "Page": page, "Pages": pages, "HasNext": total > page*limit})
	})

	mux.HandleFunc("/admin/npr/api/summary", func(w http.ResponseWriter, r *http.Request) {
		p, ok := guard(w, r)
		if !ok {
			return
		}
		tab := r.URL.Query().Get("tab")
		if _, ok := nprTabs[tab]; !ok {
			tab = "margin"
		}
		ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second)
		defer cancel()
		var total, below1, below0, calls, inf, warn, okc, tabM, tabD int64
		err := p.QueryRow(ctx, `WITH per_client AS (SELECT DISTINCT ON (client_code) client_code, uds, uds_flag, zone, init_margin, min_margin, portfolio_value FROM npr.v_current WHERE `+nprTabs[tab]+` ORDER BY client_code, (COALESCE(abs(rcv1),0) + COALESCE(abs(rcv2),0)) DESC)
			SELECT count(*), count(*) FILTER (WHERE uds IS NOT NULL AND uds < 1), count(*) FILTER (WHERE uds IS NOT NULL AND uds < 0), count(*) FILTER (WHERE zone='CALL'), count(*) FILTER (WHERE zone='INF'), count(*) FILTER (WHERE zone='WARN'), count(*) FILTER (WHERE zone='OK') FROM per_client`).
			Scan(&total, &below1, &below0, &calls, &inf, &warn, &okc)
		if err == nil {
			err = p.QueryRow(ctx, `SELECT count(DISTINCT client_code) FILTER (WHERE `+nprTabs["margin"]+`), count(DISTINCT client_code) FILTER (WHERE `+nprTabs["debt"]+`) FROM npr.v_current`).Scan(&tabM, &tabD)
		}
		if err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{"total": total, "below_one": below1, "below_zero": below0, "calls": calls, "inf": inf, "warn": warn, "ok": okc, "tab_margin": tabM, "tab_debt": tabD, "tab": tab})
	})

	mux.HandleFunc("/admin/npr/api/freshness", func(w http.ResponseWriter, r *http.Request) {
		p, ok := guard(w, r)
		if !ok {
			return
		}
		ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second)
		defer cancel()
		var lastTick *time.Time
		var ticks int64
		_ = p.QueryRow(ctx, "SELECT last_tick, tick_count FROM npr.heartbeat WHERE id=1").Scan(&lastTick, &ticks)
		var lastTS *time.Time
		var total, hour, day int64
		_ = p.QueryRow(ctx, `SELECT max(ts), count(*), count(*) FILTER (WHERE ts >= NOW() - INTERVAL '1 hour'), count(*) FILTER (WHERE ts >= NOW() - INTERVAL '24 hours') FROM npr.history`).Scan(&lastTS, &total, &hour, &day)
		out := map[string]any{"last_tick": nil, "tick_count": ticks, "last_event_ts": nil, "total_rows": total, "events_last_hour": hour, "events_last_24h": day, "server_now": time.Now().Format(time.RFC3339)}
		if lastTick != nil {
			out["last_tick"] = lastTick.Format(time.RFC3339)
		}
		if lastTS != nil {
			out["last_event_ts"] = lastTS.Format(time.RFC3339)
		}
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(out)
	})

	mux.HandleFunc("/admin/npr/c/", func(w http.ResponseWriter, r *http.Request) {
		p, ok := guard(w, r)
		if !ok {
			return
		}
		rest := strings.TrimPrefix(r.URL.Path, "/admin/npr/c/")
		code := rest
		sub := ""
		if i := strings.Index(rest, "/"); i >= 0 {
			code, sub = rest[:i], rest[i+1:]
		}
		if code == "" {
			http.NotFound(w, r)
			return
		}
		hours, _ := strconv.Atoi(r.URL.Query().Get("hours"))
		tf, tt, custom := nprParsePeriod(hours, r.URL.Query().Get("dt_from"), r.URL.Query().Get("dt_to"))
		if hours <= 0 {
			hours = 24
		}
		ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
		defer cancel()
		switch sub {
		case "series":
			rows, err := p.Query(ctx, "SELECT ts, rcv1::float8, rcv2::float8, uds::float8 FROM npr.v_npr WHERE client_code=$1 AND ts BETWEEN $2 AND $3 ORDER BY ts", code, tf, tt)
			if err != nil {
				http.Error(w, err.Error(), 500)
				return
			}
			defer rows.Close()
			labels, r1, r2, u := []string{}, []*float64{}, []*float64{}, []*float64{}
			for rows.Next() {
				var ts time.Time
				var a, b, c *float64
				if rows.Scan(&ts, &a, &b, &c) == nil {
					labels = append(labels, ts.Format(time.RFC3339))
					r1, r2, u = append(r1, a), append(r2, b), append(u, c)
				}
			}
			w.Header().Set("Content-Type", "application/json")
			_ = json.NewEncoder(w).Encode(map[string]any{"labels": labels, "rcv1": r1, "rcv2": r2, "uds": u})
			return
		case "export.csv":
			rows, err := p.Query(ctx, `SELECT h.ts, h.firm_id, h.client_code, COALESCE(h.margin_type,''), h.rcv1::float8, h.rcv2::float8,
				CASE WHEN h.funds_level >= 9.99 THEN NULL ELSE h.funds_level::float8 END, CASE WHEN h.funds_level >= 9.99 THEN 'INF' WHEN h.funds_level <= -9.99 THEN 'CALL' ELSE '' END,
				h.portfolio_value::float8, h.init_margin::float8, h.min_margin::float8, h.corrected_margin::float8, GREATEST(-h.rcv1, 0)::float8, h.status, h.risk_level
				FROM npr.history h WHERE h.client_code=$1 AND h.ts BETWEEN $2 AND $3 ORDER BY h.ts`, code, tf, tt)
			if err != nil {
				http.Error(w, err.Error(), 500)
				return
			}
			defer rows.Close()
			w.Header().Set("Content-Type", "text/csv; charset=utf-8")
			w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="npr_%s_%s-%s.csv"`, code, tf.Format("20060102"), tt.Format("20060102")))
			w.Write([]byte{0xEF, 0xBB, 0xBF})
			cw := csv.NewWriter(w)
			cw.Comma = ';'
			cw.Write([]string{"Время (UTC)", "Время (МСК)", "Фирма", "Код клиента", "Тип клиента", "НПР1", "НПР2", "УДС", "Флаг УДС", "Стоимость портфеля", "Начальная маржа", "Минимальная маржа", "Скорректированная маржа", "Маржинальное требование", "Статус", "Уровень риска"})
			f := func(v *float64) string {
				if v == nil {
					return ""
				}
				return strings.ReplaceAll(strconv.FormatFloat(*v, 'f', 4, 64), ".", ",")
			}
			for rows.Next() {
				var ts time.Time
				var firm, cc, mt, flag string
				var a, b, c, pv, im, mm, cm, mr *float64
				var st, rl *int
				if rows.Scan(&ts, &firm, &cc, &mt, &a, &b, &c, &flag, &pv, &im, &mm, &cm, &mr, &st, &rl) != nil {
					continue
				}
				cw.Write([]string{ts.UTC().Format("2006-01-02 15:04:05"), ts.In(nprMSK).Format("2006-01-02 15:04:05"), firm, cc, mt, f(a), f(b), f(c), flag, f(pv), f(im), f(mm), f(cm), f(mr), nprI(st), nprI(rl)})
			}
			cw.Flush()
			return
		case "":
		default:
			http.NotFound(w, r)
			return
		}
		rows, err := p.Query(ctx, `SELECT `+nprCols+`, 0 FROM npr.v_current WHERE client_code=$1 ORDER BY (COALESCE(abs(rcv1),0) + COALESCE(abs(rcv2),0)) DESC LIMIT 1`, code)
		if err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		cur, _, err := scanNPR(rows)
		rows.Close()
		if err != nil || len(cur) == 0 {
			http.Error(w, "Клиент "+code+" не найден в текущем состоянии QUIK", http.StatusNotFound)
			return
		}
		hrows, err := p.Query(ctx, `SELECT client_code, margin_type, uds::float8, uds_flag, rcv1::float8, rcv2::float8, margin_requirement::float8, portfolio_value::float8, init_margin::float8, min_margin::float8, NULL::float8, NULL::int, NULL::int, ts, NULL::text, 0 FROM npr.v_npr WHERE client_code=$1 AND ts BETWEEN $2 AND $3 ORDER BY ts DESC LIMIT 200`, code, tf, tt)
		if err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		hist, _, _ := scanNPR(hrows)
		hrows.Close()
		data := struct {
			page
			C            nprRow
			History      []nprRow
			Hours        int
			DtFrom, DtTo string
			TF, TT       time.Time
			Custom       bool
			Periods      []struct {
				H int
				T string
			}
		}{nowPage("Клиент "+code+" — НПР/УДС", "npr"), cur[0], hist, hours, r.URL.Query().Get("dt_from"), r.URL.Query().Get("dt_to"), tf, tt, custom,
			[]struct {
				H int
				T string
			}{{1, "за 1 час"}, {6, "за 6 часов"}, {24, "за сутки"}, {168, "за 7 дней"}, {720, "за 30 дней"}}}
		render(w, a.nprClient, data)
	})
	log.Printf("npr: модуль зарегистрирован (включён: %v)", nprAvailable(rc))
}

// nprTest — проверка подключения для настроек.
func nprTest(rc *RuntimeConfig) (string, error) {
	p, err := nprDB(rc)
	if err != nil {
		return "", err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	var clients, hist int64
	var tick *time.Time
	if err := p.QueryRow(ctx, "SELECT count(DISTINCT client_code) FROM npr.v_current").Scan(&clients); err != nil {
		return "", fmt.Errorf("npr.v_current: %v", err)
	}
	_ = p.QueryRow(ctx, "SELECT count(*) FROM npr.history").Scan(&hist)
	_ = p.QueryRow(ctx, "SELECT last_tick FROM npr.heartbeat WHERE id=1").Scan(&tick)
	age := "нет пульса"
	if tick != nil {
		age = fmt.Sprintf("пульс экспортёра %s назад", time.Since(*tick).Round(time.Second))
	}
	return fmt.Sprintf("клиентов в текущем состоянии %d, записей истории %d, %s", clients, hist, age), nil
}

func (h *setupHandlers) saveNPR(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method", http.StatusMethodNotAllowed)
		return
	}
	s := NPRSettings{Enabled: r.FormValue("enabled") != "", DSN: strings.TrimSpace(r.FormValue("dsn"))}
	if err := h.rc.UpdateNPR(s); err != nil {
		setupFlash(w, r, "Модуль НПР: ошибка сохранения: "+err.Error())
		return
	}
	if s.Enabled {
		setupFlash(w, r, "Модуль НПР / УДС включён")
	} else {
		setupFlash(w, r, "Модуль НПР / УДС выключен")
	}
}

func (h *setupHandlers) testNPR(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method", http.StatusMethodNotAllowed)
		return
	}
	if dsn := strings.TrimSpace(r.FormValue("dsn")); dsn != "" {
		s := h.rc.Snapshot().NPR
		if s.DSN != dsn {
			s.DSN = dsn
			_ = h.rc.UpdateNPR(s)
		}
	}
	msg, err := nprTest(h.rc)
	if err != nil {
		setupFlash(w, r, "Модуль НПР: подключение НЕ удалось — "+err.Error())
		return
	}
	setupFlash(w, r, "Модуль НПР: подключение OK — "+msg)
}