DEPO: строчные события по депо-поручениям (частичное исполнение пакета)
По ответам программиста Fancy (В. Кокин, 21.09.2026) на 8 вопросов из docs/fancy-events-partial-m2m.md. Одно депо-поручение (MF010) = одна бумага пакета, поэтому подача и исход теперь пишутся в DEPO ПО СТРОКАМ с REFERENCE и сразу по MS010 этой строки, не дожидаясь остальных: пакет может исполниться частично (Сбер — да, Т-Технологии — отказ). События на весь GUID (пустой REFERENCE) больше не пишем — переход одновременно с bj2depo. - BJ_ORDER_SENT / BJ_ORDER_ACCEPTED / BJ_EXECUTED / BJ_ORDER_REJECTED / BJ_ORDER_LATE / BJ_CANCELLED — по строке (depoOnOrderLineSent/Result); contr_d_id кладём в COMMENT (джобе связка не нужна — по REFERENCE найдёт). - OrderWatch.Lines — состояние по строкам (contr_d_id → REFERENCE/ISIN/MS010/ отказ/истечение); объект закрыт, когда MS010 по всем строкам; 30 дней — BJ_CANCELLED по каждой незакрытой строке, QUIK-блок снимается по её ISIN. - Входящий 16/1 хранит ВСЕ contr_d_id (OrderLines) — раньше только первый (дефект «входящие отслеживают только первый contr_d_id»). Исходящий 16: REFERENCE по ISIN из бумаг M2MTR (outgoingLineRefs, кэш по GUID; svc в deps). - Карточка: «исполнено частично: N из M» (StatusRu), бейдж warn — как статус «частично» в DEPO. Backfill и эмуляция — тоже по строкам. - Попутно: EXEC процедуры обработчика DEPO — через execProc с проверкой имени (sqlident.go), чтобы значение из настроек не стало произвольным T-SQL. Тесты: TestOrderWatchLinesPartial, TestIncomingOrderLines, TestOutgoingLineRefsNoRepo; весь пакет зелёный. РАЗВЁРТЫВАНИЕ — только одновременно с bj2depo (п. 8 ответов Fancy). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Q4YVo4UMopqT5DzicqpPjL
This commit is contained in:
zuevav
@@ -14,8 +14,11 @@ import (
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2mcore"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -28,10 +31,56 @@ type OrderWatch struct {
|
||||
GUID string `json:"guid"`
|
||||
Direction string `json:"direction"` // out (16) | in (16/1)
|
||||
OrderAt time.Time `json:"order_at"`
|
||||
MS010At time.Time `json:"ms010_at,omitempty"` // все строки получили MS010
|
||||
Rejected bool `json:"rejected,omitempty"` // хотя бы одна строка отвергнута
|
||||
LateAt time.Time `json:"late_at,omitempty"` // когда зафиксировали просрочку 5 мин
|
||||
ExpiredAt time.Time `json:"expired_at,omitempty"` // все незакрытые строки истекли (30 дней)
|
||||
// Lines — состояние по каждой строке пакета (одно депо-поручение = одна бумага),
|
||||
// ключ — contr_d_id. По договорённости с Fancy (18.09.2026) исходы пишутся в DEPO
|
||||
// по строкам с REFERENCE и сразу по MS010 этой строки, не дожидаясь остальных:
|
||||
// пакет может исполниться частично (Сбер — да, Т-Технологии — отказ).
|
||||
Lines map[string]*OrderLineWatch `json:"lines,omitempty"`
|
||||
}
|
||||
|
||||
// OrderLineWatch — одна строка поручения: какая бумага (REFERENCE/ISIN) и её исход.
|
||||
type OrderLineWatch struct {
|
||||
Ref string `json:"ref,omitempty"` // REFERENCE строки в FTV_M2M_LINES
|
||||
ISIN string `json:"isin,omitempty"`
|
||||
MS010At time.Time `json:"ms010_at,omitempty"`
|
||||
Rejected bool `json:"rejected,omitempty"`
|
||||
LateAt time.Time `json:"late_at,omitempty"` // когда зафиксировали просрочку 5 мин
|
||||
ExpiredAt time.Time `json:"expired_at,omitempty"` // когда истекли 30 дней
|
||||
ExpiredAt time.Time `json:"expired_at,omitempty"`
|
||||
}
|
||||
|
||||
// orderLine — строка поручения для наблюдателя: contr_d_id ↔ бумага (REFERENCE, ISIN).
|
||||
type orderLine struct {
|
||||
ContrDID string
|
||||
Ref string
|
||||
ISIN string
|
||||
}
|
||||
|
||||
// lineDone — строка в финальном состоянии (MS010 получен или срок истёк).
|
||||
func (l *OrderLineWatch) lineDone() bool {
|
||||
return l != nil && (!l.MS010At.IsZero() || !l.ExpiredAt.IsZero())
|
||||
}
|
||||
|
||||
// linesSummary — сколько строк закрыто MS010 / отвергнуто / истекло из общего числа.
|
||||
// Нужно для частичного исполнения: пакет может быть исполнен не целиком.
|
||||
func (w *OrderWatch) linesSummary() (total, executed, rejected, expired int) {
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
for _, l := range w.Lines {
|
||||
total++
|
||||
switch {
|
||||
case !l.MS010At.IsZero() && l.Rejected:
|
||||
rejected++
|
||||
case !l.MS010At.IsZero():
|
||||
executed++
|
||||
case !l.ExpiredAt.IsZero():
|
||||
expired++
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Состояние по-русски для карточек.
|
||||
@@ -39,6 +88,20 @@ func (w *OrderWatch) StatusRu() string {
|
||||
if w == nil {
|
||||
return ""
|
||||
}
|
||||
// Частичное исполнение: часть строк закрыта, часть ещё ждёт / отвергнута / истекла.
|
||||
if total, executed, rejected, expired := w.linesSummary(); total > 1 && executed > 0 && executed < total {
|
||||
s := fmt.Sprintf("исполнено частично: %d из %d бумаг", executed, total)
|
||||
if rejected > 0 {
|
||||
s += fmt.Sprintf(", отвергнуто %d", rejected)
|
||||
}
|
||||
if expired > 0 {
|
||||
s += fmt.Sprintf(", истёк срок по %d", expired)
|
||||
}
|
||||
if executed+rejected+expired < total {
|
||||
s += fmt.Sprintf(", ожидаем MS010 по %d", total-executed-rejected-expired)
|
||||
}
|
||||
return s
|
||||
}
|
||||
switch {
|
||||
case !w.MS010At.IsZero() && w.Rejected:
|
||||
return "НРД отверг поручение (MS010 " + w.MS010At.Local().Format("02.01 15:04") + ")"
|
||||
@@ -58,6 +121,12 @@ func (w *OrderWatch) StatusRu() string {
|
||||
}
|
||||
|
||||
func (w *OrderWatch) Kind() string {
|
||||
if w != nil {
|
||||
// частично исполнено — предупреждение, а не «отказ» и не «ок»
|
||||
if total, executed, _, _ := w.linesSummary(); total > 1 && executed > 0 && executed < total {
|
||||
return "warn"
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case w == nil:
|
||||
return ""
|
||||
@@ -133,9 +202,57 @@ func (s *orderWatchStore) update(guid string, fn func(w *OrderWatch)) *OrderWatc
|
||||
|
||||
var orderWatchStarted = time.Now()
|
||||
|
||||
// orderWatchCheck — один объект: подано когда, ответы НРД → фиксируем MS010/просрочку/истечение.
|
||||
func orderWatchCheck(ctx context.Context, rc *RuntimeConfig, guid, direction string, orderAt time.Time, resps []orderResponse, label string) {
|
||||
if guid == "" || orderAt.IsZero() {
|
||||
// outgoingLineRefsCache — REFERENCE по ISIN для исходящей сделки: берётся из
|
||||
// M2MTR-запроса в репозитории; после отправки не меняется, поэтому кэшируем по GUID.
|
||||
var outgoingLineRefsCache = struct {
|
||||
mu sync.Mutex
|
||||
m map[string]map[string]string // guid → ISIN → ref
|
||||
}{m: map[string]map[string]string{}}
|
||||
|
||||
// outgoingLineRefs — строки исходящего поручения 16: contr_d_id ↔ ISIN ↔ REFERENCE.
|
||||
// Один черновик MF010 = одна бумага; REFERENCE берём из бумаг M2MTR по ISIN.
|
||||
func outgoingLineRefs(ctx context.Context, svc *Service, od *OutgoingDeal) []orderLine {
|
||||
if od == nil {
|
||||
return nil
|
||||
}
|
||||
outgoingLineRefsCache.mu.Lock()
|
||||
byISIN, ok := outgoingLineRefsCache.m[od.GUID]
|
||||
outgoingLineRefsCache.mu.Unlock()
|
||||
if !ok && svc != nil {
|
||||
byISIN = map[string]string{}
|
||||
if deals, err := svc.Repo().List(ctx, m2mcore.Filter{Limit: 500}); err == nil {
|
||||
for _, d := range deals {
|
||||
if d == nil || d.Request == nil || !strings.EqualFold(string(d.GUID), od.GUID) {
|
||||
continue
|
||||
}
|
||||
for _, sec := range d.Request.Data.TransferredSecurities.Securities {
|
||||
byISIN[strings.ToUpper(quikSecISIN(sec))] = string(sec.ReferenceID)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(byISIN) > 0 {
|
||||
outgoingLineRefsCache.mu.Lock()
|
||||
outgoingLineRefsCache.m[od.GUID] = byISIN
|
||||
outgoingLineRefsCache.mu.Unlock()
|
||||
}
|
||||
}
|
||||
var out []orderLine
|
||||
for _, sec := range od.Secs {
|
||||
if sec.ContrDID == "" {
|
||||
continue
|
||||
}
|
||||
isin := strings.ToUpper(sec.ISIN)
|
||||
out = append(out, orderLine{ContrDID: sec.ContrDID, Ref: byISIN[isin], ISIN: isin})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// orderWatchCheck — один объект (перевод): подано когда, ответы НРД ПО КАЖДОЙ СТРОКЕ →
|
||||
// фиксируем MS010/просрочку/истечение и пишем строчные события в DEPO сразу по
|
||||
// MS010 этой бумаги, не дожидаясь остальных (частичное исполнение пакета).
|
||||
func orderWatchCheck(ctx context.Context, rc *RuntimeConfig, guid, direction string, orderAt time.Time, lines []orderLine, label string) {
|
||||
if guid == "" || orderAt.IsZero() || len(lines) == 0 {
|
||||
return
|
||||
}
|
||||
w := globalOrderWatch.Get(guid)
|
||||
@@ -143,60 +260,120 @@ func orderWatchCheck(ctx context.Context, rc *RuntimeConfig, guid, direction str
|
||||
w = globalOrderWatch.update(guid, func(w *OrderWatch) { w.Direction, w.OrderAt = direction, orderAt })
|
||||
}
|
||||
if !w.MS010At.IsZero() || !w.ExpiredAt.IsZero() {
|
||||
return // финальное состояние уже зафиксировано
|
||||
}
|
||||
if done, rejected, text := orderRespVerdict(resps); done {
|
||||
at := time.Now()
|
||||
for _, r := range resps {
|
||||
if r.Done {
|
||||
if t, err := time.ParseInLocation("02.01.2006 15:04:05", r.Time, time.Local); err == nil {
|
||||
at = t
|
||||
}
|
||||
}
|
||||
}
|
||||
globalOrderWatch.update(guid, func(w *OrderWatch) { w.MS010At, w.Rejected = at, rejected })
|
||||
if rejected {
|
||||
go depoOnOrderResult(rc, guid, "rejected", text)
|
||||
} else {
|
||||
go depoOnOrderResult(rc, guid, "accepted", "")
|
||||
}
|
||||
return
|
||||
return // финальное состояние уже зафиксировано по всем строкам
|
||||
}
|
||||
now := time.Now()
|
||||
fresh := orderAt.After(orderWatchStarted) // по старым поручениям при первом запуске не шумим
|
||||
|
||||
// 1) исходы по строкам: у каждой бумаги свой MS010
|
||||
for _, ln := range lines {
|
||||
st := w.Lines[ln.ContrDID]
|
||||
if st.lineDone() {
|
||||
continue
|
||||
}
|
||||
resps := findOrderResponses(ln.ContrDID)
|
||||
done, rejected, text := orderRespVerdict(resps)
|
||||
if !done {
|
||||
continue
|
||||
}
|
||||
at := orderRespAt(resps)
|
||||
globalOrderWatch.update(guid, func(w *OrderWatch) {
|
||||
if w.Lines == nil {
|
||||
w.Lines = map[string]*OrderLineWatch{}
|
||||
}
|
||||
w.Lines[ln.ContrDID] = &OrderLineWatch{Ref: ln.Ref, ISIN: ln.ISIN, MS010At: at, Rejected: rejected}
|
||||
if rejected {
|
||||
w.Rejected = true
|
||||
}
|
||||
})
|
||||
if rejected {
|
||||
go depoOnOrderLineResult(rc, guid, ln.Ref, "rejected", text)
|
||||
} else {
|
||||
go depoOnOrderLineResult(rc, guid, ln.Ref, "accepted", "")
|
||||
}
|
||||
}
|
||||
w = globalOrderWatch.Get(guid)
|
||||
|
||||
// 2) все строки закрыты MS010 → финальное состояние объекта
|
||||
if allLinesMS010(w, lines) {
|
||||
latest := time.Time{}
|
||||
for _, ln := range lines {
|
||||
if st := w.Lines[ln.ContrDID]; st != nil && st.MS010At.After(latest) {
|
||||
latest = st.MS010At
|
||||
}
|
||||
}
|
||||
globalOrderWatch.update(guid, func(w *OrderWatch) { w.MS010At = latest })
|
||||
return
|
||||
}
|
||||
|
||||
// 3) просрочка 5 минут — один сигнал на объект, событие по каждой незакрытой строке
|
||||
if w.LateAt.IsZero() && now.Sub(orderAt) > orderMS010Within {
|
||||
globalOrderWatch.update(guid, func(w *OrderWatch) { w.LateAt = now })
|
||||
go depoOnOrderResult(rc, guid, "late", "")
|
||||
for _, ln := range lines {
|
||||
if !w.Lines[ln.ContrDID].lineDone() {
|
||||
go depoOnOrderLineResult(rc, guid, ln.Ref, "late", "")
|
||||
}
|
||||
}
|
||||
if fresh {
|
||||
raiseSystemAlert(rc, "НРД: депо-поручение не исполнено в срок 5 минут",
|
||||
fmt.Sprintf("%s (перевод %s): подано %s, отчёт MS010 не получен. Ожидаем исполнения до %s, после этого поручение считается отменённым.",
|
||||
label, shortGUID(guid), orderAt.Local().Format("02.01.2006 15:04"), orderAt.Add(orderExecWithin).Local().Format("02.01.2006")))
|
||||
}
|
||||
}
|
||||
if w.ExpiredAt.IsZero() && now.Sub(orderAt) > orderExecWithin {
|
||||
globalOrderWatch.update(guid, func(w *OrderWatch) { w.ExpiredAt = now })
|
||||
go depoOnOrderResult(rc, guid, "expired", "")
|
||||
raiseSystemAlert(rc, "НРД: истёк срок 30 дней по депо-поручению — считается отменённым",
|
||||
fmt.Sprintf("%s (перевод %s): подано %s, отчёт MS010 так и не получен. Поручение считается отменённым; проверьте перевод в НРД и позицию клиента.",
|
||||
label, shortGUID(guid), orderAt.Local().Format("02.01.2006 15:04")))
|
||||
if direction == "out" {
|
||||
for _, l := range globalQuikLinks.ForGUID(guid) {
|
||||
if l.State == "registered" {
|
||||
_, _ = QuikOutDeny(ctx, rc, guid, l.ISIN, "Истёк срок исполнения поручения в НРД (30 дней) — перевод отменён")
|
||||
|
||||
// 4) 30 дней — по каждой незакрытой строке BJ_CANCELLED с REFERENCE; QUIK-блокировка снимается по ISIN
|
||||
if now.Sub(orderAt) > orderExecWithin {
|
||||
expiredAny := false
|
||||
for _, ln := range lines {
|
||||
if w.Lines[ln.ContrDID].lineDone() {
|
||||
continue
|
||||
}
|
||||
expiredAny = true
|
||||
globalOrderWatch.update(guid, func(w *OrderWatch) {
|
||||
if w.Lines == nil {
|
||||
w.Lines = map[string]*OrderLineWatch{}
|
||||
}
|
||||
w.Lines[ln.ContrDID] = &OrderLineWatch{Ref: ln.Ref, ISIN: ln.ISIN, ExpiredAt: now}
|
||||
})
|
||||
go depoOnOrderLineResult(rc, guid, ln.Ref, "expired", "")
|
||||
if direction == "out" {
|
||||
for _, l := range globalQuikLinks.ForGUID(guid) {
|
||||
if l.State == "registered" && (ln.ISIN == "" || strings.EqualFold(l.ISIN, ln.ISIN)) {
|
||||
_, _ = QuikOutDeny(ctx, rc, guid, l.ISIN, "Истёк срок исполнения поручения в НРД (30 дней) — перевод отменён")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if expiredAny {
|
||||
globalOrderWatch.update(guid, func(w *OrderWatch) { w.ExpiredAt = now })
|
||||
raiseSystemAlert(rc, "НРД: истёк срок 30 дней по депо-поручению — считается отменённым",
|
||||
fmt.Sprintf("%s (перевод %s): подано %s, отчёт MS010 так и не получен. Поручение считается отменённым; проверьте перевод в НРД и позицию клиента.",
|
||||
label, shortGUID(guid), orderAt.Local().Format("02.01.2006 15:04")))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func orderWatchTick(ctx context.Context, rc *RuntimeConfig) {
|
||||
// allLinesMS010 — по всем строкам получен MS010.
|
||||
func allLinesMS010(w *OrderWatch, lines []orderLine) bool {
|
||||
if w == nil || len(lines) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, ln := range lines {
|
||||
st := w.Lines[ln.ContrDID]
|
||||
if st == nil || st.MS010At.IsZero() {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func orderWatchTick(ctx context.Context, rc *RuntimeConfig, svc *Service) {
|
||||
if globalOutgoingStore != nil {
|
||||
for _, od := range scanForm16Drafts(globalOutgoingStore) {
|
||||
if !od.Submitted {
|
||||
continue
|
||||
}
|
||||
resps, _ := outgoingResponses(od.ContrDIDs)
|
||||
orderWatchCheck(ctx, rc, od.GUID, "out", od.OrderAt, resps, "Поручение 16 (списание)")
|
||||
orderWatchCheck(ctx, rc, od.GUID, "out", od.OrderAt, outgoingLineRefs(ctx, svc, od), "Поручение 16 (списание)")
|
||||
}
|
||||
}
|
||||
if globalIncomingStore != nil {
|
||||
@@ -204,11 +381,30 @@ func orderWatchTick(ctx context.Context, rc *RuntimeConfig) {
|
||||
if !it.OrderSubmitted {
|
||||
continue
|
||||
}
|
||||
orderWatchCheck(ctx, rc, it.GUID, "in", it.OrderAt, findOrderResponses(it.OrderContrDID), "Поручение 16/1 (зачисление)")
|
||||
orderWatchCheck(ctx, rc, it.GUID, "in", it.OrderAt, incomingOrderLines(it), "Поручение 16/1 (зачисление)")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// incomingOrderLines — строки поданных 16/1. Старые записи (до строчных событий) знали
|
||||
// только первый contr_d_id — для них одна строка без REFERENCE.
|
||||
func incomingOrderLines(it *IncomingTransfer) []orderLine {
|
||||
if it == nil {
|
||||
return nil
|
||||
}
|
||||
if len(it.OrderLines) > 0 {
|
||||
out := make([]orderLine, 0, len(it.OrderLines))
|
||||
for _, l := range it.OrderLines {
|
||||
out = append(out, orderLine{ContrDID: l.ContrDID, Ref: l.Ref, ISIN: l.ISIN})
|
||||
}
|
||||
return out
|
||||
}
|
||||
if it.OrderContrDID != "" {
|
||||
return []orderLine{{ContrDID: it.OrderContrDID}}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// orderWatchLoop — фоновый контроль сроков (раз в минуту).
|
||||
func (s *Server) orderWatchLoop(ctx context.Context) {
|
||||
ticker := time.NewTicker(time.Minute)
|
||||
@@ -225,7 +421,7 @@ func (s *Server) orderWatchLoop(ctx context.Context) {
|
||||
log.Printf("order-watch: panic: %v", r)
|
||||
}
|
||||
}()
|
||||
orderWatchTick(ctx, s.rc)
|
||||
orderWatchTick(ctx, s.rc, s.svc)
|
||||
}()
|
||||
healthBeat("order-watch")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user