НПР / УДС — отдельная служба bj-npr

Модуль вынесен в modules/npr + cmd/bj-npr (127.0.0.1:8083), обновляется
независимо от ядра. Служба читает подключение из setup.json и принимает
запросы только со служебным ключом модулей (/var/lib/bj/.bj/module.key,
создаёт bj-server). Полные страницы отдаются фрагментом (X-BJ-Layout: page).

bj-server хранит настройки и проверку подключения, проксирует /admin/npr
после входа, прав и лицензии и встраивает страницы модуля в общий макет
(admin.modulePage); если служба не отвечает — понятная страница.
Unit bj-npr.service встроен в установщик, установщик включает службы
установленных модулей; build-release.sh собирает bj-npr.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
zuevav
2026-09-16 01:00:31 +03:00
parent d8dee07213
commit 5527a464ca
15 changed files with 872 additions and 467 deletions
+601
View File
@@ -0,0 +1,601 @@
// 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"
"crypto/subtle"
"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"
)
//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
}
// requireKey — принимать запросы только от bj-server (кроме /healthz).
func (s *Server) requireKey(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/healthz" {
next.ServeHTTP(w, r)
return
}
key, err := os.ReadFile(s.cfg.KeyPath)
want := strings.TrimSpace(string(key))
got := r.Header.Get("X-BJ-Module-Key")
if err != nil || want == "" || subtle.ConstantTimeCompare([]byte(want), []byte(got)) != 1 {
http.Error(w, "модуль принимает запросы только от bj-server", http.StatusForbidden)
return
}
next.ServeHTTP(w, r)
})
}
// 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 s.requireKey(mux)
}