// Package npr — модуль «НПР / УДС»: риск-монитор клиентов маржинальной торговли по базе экспорта // QUIK (quik_export, схема npr), логином только для чтения. // // Работает отдельной службой bj-npr (cmd/bj-npr) и обновляется независимо от ядра. bj-server // проверяет вход, права и лицензию и проксирует ей раздел /admin/npr со служебным ключом // (X-BJ-Module-Key). Полные страницы служба отдаёт фрагментом с заголовком X-BJ-Layout: page, а // bj-server встраивает их в свой макет с меню; строки таблицы, JSON, CSV и статика идут как есть. // Подключение к базе служба читает из настроек bj (раздел npr в setup.json). package npr import ( "bytes" "context" "embed" "encoding/csv" "encoding/json" "fmt" "html/template" "io/fs" "math" "net/http" "net/url" "os" "strconv" "strings" "sync" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "git.zetit.ru/zuevav/Bridge-and-Join-s/modules/modkit" ) //go:embed web/*.html var tplFS embed.FS //go:embed web/static/* var staticFS embed.FS // Config — где служба берёт настройки и ключ. type Config struct { SetupPath string // setup.json bj: раздел npr {enabled, dsn} KeyPath string // ключ модуля; bj-server передаёт его в X-BJ-Module-Key } // Server — служба модуля. type Server struct { cfg Config list, client, rows *template.Template mu sync.Mutex pool *pgxpool.Pool poolDSN string } var msk = time.FixedZone("MSK", 3*3600) // New — служба с разобранными шаблонами. func New(cfg Config) (*Server, error) { funcs := template.FuncMap{ "nprF": fmtF, "deref": func(v *float64) float64 { if v == nil { return 0 } return *v }, "add": func(a, b int) int { return a + b }, "list": func(v ...string) []string { return v }, "has": func(l []string, v string) bool { for _, x := range l { if x == v { return true } } return false }, "dict": func(kv ...string) map[string]string { m := map[string]string{} for i := 0; i+1 < len(kv); i += 2 { m[kv[i]] = kv[i+1] } return m }, } parse := func(files ...string) (*template.Template, error) { return template.New("").Funcs(funcs).ParseFS(tplFS, files...) } s := &Server{cfg: cfg} var err error if s.list, err = parse("web/css.html", "web/list.html"); err != nil { return nil, err } if s.client, err = parse("web/css.html", "web/client.html"); err != nil { return nil, err } if s.rows, err = parse("web/rows.html"); err != nil { return nil, err } return s, nil } // ---- настройки, ключ, база ---- func (s *Server) settings() (enabled bool, dsn string) { b, err := os.ReadFile(s.cfg.SetupPath) if err != nil { return false, "" } var st struct { NPR struct { Enabled bool `json:"enabled"` DSN string `json:"dsn"` } `json:"npr"` } if json.Unmarshal(b, &st) != nil { return false, "" } return st.NPR.Enabled, strings.TrimSpace(st.NPR.DSN) } func (s *Server) db() (*pgxpool.Pool, error) { enabled, dsn := s.settings() if !enabled || dsn == "" { return nil, fmt.Errorf("модуль не включён или не задано подключение к базе экспорта QUIK (Настройка → НПР / УДС)") } s.mu.Lock() defer s.mu.Unlock() if s.pool != nil && s.poolDSN == dsn { return s.pool, nil } if s.pool != nil { s.pool.Close() s.pool = 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 } s.pool, s.poolDSN = p, dsn return p, nil } // page — полная страница фрагментом для макета bj-server. func (s *Server) page(w http.ResponseWriter, t *template.Template, name, title string, data any) { var buf bytes.Buffer if err := t.ExecuteTemplate(&buf, name, data); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } h := w.Header() h.Set("Content-Type", "text/html; charset=utf-8") h.Set("Cache-Control", "no-store") h.Set("X-BJ-Layout", "page") h.Set("X-BJ-Title", url.PathEscape(title)) _, _ = w.Write(buf.Bytes()) } // ---- данные ---- var sortOrders = 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 tabs = 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", } // Row — клиент в текущем состоянии или запись истории. type Row 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 Row) 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 Row) 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 Row) BadgeClass() string { c, _ := r.Badge(); return c } func (r Row) BadgeText() string { _, t := r.Badge(); return t } func fmtF(v *float64) string { if v == nil || math.IsNaN(*v) { return "—" } return strings.ReplaceAll(strconv.FormatFloat(*v, 'f', 2, 64), ".", ",") } func fmtI(v *int) string { if v == nil { return "" } return strconv.Itoa(*v) } func parseF(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 where(q string, mt []string, umn, umx *float64, onlyCalls bool, tab string) (string, []any) { var cl []string var params []any cl = append(cl, "("+tabs[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 scanRows(rows pgx.Rows) ([]Row, int, error) { var out []Row total := 0 for rows.Next() { var r Row var mt, fl, 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 cols = `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 parsePeriod(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, msk); 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 } // ---- маршруты ---- // Handler — все маршруты модуля; без служебного ключа отвечает 403. func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("ok")) }) sub, _ := fs.Sub(staticFS, "web/static") mux.Handle("/admin/npr/static/", http.StripPrefix("/admin/npr/static/", http.FileServer(http.FS(sub)))) guard := func(w http.ResponseWriter) (*pgxpool.Pool, bool) { p, err := s.db() 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 = parseF(v.Get("umn")), parseF(v.Get("umx")) onlyCalls = v.Get("only_calls") == "true" || v.Get("only_calls") == "on" sort = v.Get("sort") if _, ok := sortOrders[sort]; !ok { sort = "uds_asc" } tab = v.Get("tab") if _, ok := tabs[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); !ok { return } q, mt, _, _, onlyCalls, sort, tab, pg := qp(r) data := struct { Q, Sort, Tab string MT []string UMN, UMX string OnlyCalls bool Page int }{q, sort, tab, mt, r.URL.Query().Get("umn"), r.URL.Query().Get("umx"), onlyCalls, pg} s.page(w, s.list, "npr_list", "НПР / УДС", data) }) mux.HandleFunc("/admin/npr/rows", func(w http.ResponseWriter, r *http.Request) { p, ok := guard(w) if !ok { return } q, mt, umn, umx, onlyCalls, sort, tab, page := qp(r) limit := 30 cond, params := where(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`, cols, cols, cond, sortOrders[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 := scanRows(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") _ = s.rows.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) if !ok { return } tab := r.URL.Query().Get("tab") if _, ok := tabs[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 `+tabs[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 `+tabs["margin"]+`), count(DISTINCT client_code) FILTER (WHERE `+tabs["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) 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) if !ok { return } rest := strings.TrimPrefix(r.URL.Path, "/admin/npr/c/") code, sub := rest, "" 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 := parsePeriod(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(msk).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), fmtI(st), fmtI(rl)}) } cw.Flush() return case "": default: http.NotFound(w, r) return } rows, err := p.Query(ctx, `SELECT `+cols+`, 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 := scanRows(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, _, _ := scanRows(hrows) hrows.Close() type period struct { H int T string } data := struct { C Row History []Row Hours int DtFrom, DtTo string TF, TT time.Time Custom bool Periods []period }{cur[0], hist, hours, r.URL.Query().Get("dt_from"), r.URL.Query().Get("dt_to"), tf, tt, custom, []period{{1, "за 1 час"}, {6, "за 6 часов"}, {24, "за сутки"}, {168, "за 7 дней"}, {720, "за 30 дней"}}} s.page(w, s.client, "npr_client", "Клиент "+code+" · НПР / УДС", data) }) return modkit.RequireKey(s.cfg.KeyPath, mux) }