chore(snapshot): sync workspace state
This commit is contained in:
mpabi
@@ -1,77 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: trade-api
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: trade-api
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: trade-api
|
||||
spec:
|
||||
imagePullSecrets:
|
||||
- name: gitea-registry
|
||||
volumes:
|
||||
- name: trade-api-wrapper
|
||||
configMap:
|
||||
name: trade-api-wrapper
|
||||
- name: trade-api-upstream
|
||||
configMap:
|
||||
name: trade-api-upstream
|
||||
containers:
|
||||
- name: api
|
||||
image: gitea.mpabi.pl/trade/trade-api:k3s-20260111095435
|
||||
imagePullPolicy: IfNotPresent
|
||||
command: ["node", "/override/wrapper.mjs"]
|
||||
ports:
|
||||
- name: http
|
||||
containerPort: 8787
|
||||
volumeMounts:
|
||||
- name: trade-api-wrapper
|
||||
mountPath: /override/wrapper.mjs
|
||||
subPath: wrapper.mjs
|
||||
readOnly: true
|
||||
- name: trade-api-upstream
|
||||
mountPath: /app/services/api/server.mjs
|
||||
subPath: server.mjs
|
||||
readOnly: true
|
||||
env:
|
||||
- name: PORT
|
||||
value: "8787"
|
||||
- name: UPSTREAM_PORT
|
||||
value: "8788"
|
||||
- name: UPSTREAM_ENTRY
|
||||
value: "/app/services/api/server.mjs"
|
||||
- name: UPSTREAM_HOST
|
||||
value: "127.0.0.1"
|
||||
- name: APP_VERSION
|
||||
value: "staging"
|
||||
- name: HASURA_GRAPHQL_URL
|
||||
value: "http://hasura:8080/v1/graphql"
|
||||
- name: HASURA_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-hasura
|
||||
key: HASURA_GRAPHQL_ADMIN_SECRET
|
||||
- name: API_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-api
|
||||
key: API_ADMIN_SECRET
|
||||
- name: CORS_ORIGIN
|
||||
value: "*"
|
||||
readinessProbe:
|
||||
httpGet:
|
||||
path: /healthz
|
||||
port: http
|
||||
initialDelaySeconds: 2
|
||||
periodSeconds: 5
|
||||
livenessProbe:
|
||||
httpGet:
|
||||
path: /healthz
|
||||
port: http
|
||||
initialDelaySeconds: 10
|
||||
periodSeconds: 10
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,12 +0,0 @@
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: trade-api
|
||||
spec:
|
||||
type: ClusterIP
|
||||
selector:
|
||||
app.kubernetes.io/name: trade-api
|
||||
ports:
|
||||
- name: http
|
||||
port: 8787
|
||||
targetPort: http
|
||||
@@ -1,783 +0,0 @@
|
||||
import crypto from 'node:crypto';
|
||||
import http from 'node:http';
|
||||
import { spawn } from 'node:child_process';
|
||||
|
||||
const WRAPPER_PORT = Number.parseInt(String(process.env.PORT || '8787'), 10);
|
||||
const UPSTREAM_PORT = Number.parseInt(String(process.env.UPSTREAM_PORT || '8788'), 10);
|
||||
const UPSTREAM_HOST = String(process.env.UPSTREAM_HOST || '127.0.0.1');
|
||||
const UPSTREAM_ENTRY = String(process.env.UPSTREAM_ENTRY || '/app/services/api/server.mjs');
|
||||
|
||||
const HASURA_URL = String(process.env.HASURA_GRAPHQL_URL || 'http://hasura:8080/v1/graphql');
|
||||
const HASURA_ADMIN_SECRET = String(process.env.HASURA_ADMIN_SECRET || '');
|
||||
const CORS_ORIGIN = String(process.env.CORS_ORIGIN || '*');
|
||||
|
||||
if (!Number.isInteger(WRAPPER_PORT) || WRAPPER_PORT <= 0) throw new Error('Invalid PORT');
|
||||
if (!Number.isInteger(UPSTREAM_PORT) || UPSTREAM_PORT <= 0) throw new Error('Invalid UPSTREAM_PORT');
|
||||
|
||||
function getIsoNow() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function sha256Hex(text) {
|
||||
return crypto.createHash('sha256').update(text, 'utf8').digest('hex');
|
||||
}
|
||||
|
||||
function getHeader(req, name) {
|
||||
const v = req.headers[String(name).toLowerCase()];
|
||||
return Array.isArray(v) ? v[0] : v;
|
||||
}
|
||||
|
||||
function readBearerToken(req) {
|
||||
const auth = getHeader(req, 'authorization');
|
||||
if (auth && typeof auth === 'string') {
|
||||
const m = auth.match(/^Bearer\s+(.+)$/i);
|
||||
if (m && m[1]) return m[1].trim();
|
||||
}
|
||||
const apiKey = getHeader(req, 'x-api-key');
|
||||
if (apiKey && typeof apiKey === 'string' && apiKey.trim()) return apiKey.trim();
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function withCors(res) {
|
||||
res.setHeader('access-control-allow-origin', CORS_ORIGIN);
|
||||
res.setHeader('access-control-allow-methods', 'GET,POST,OPTIONS');
|
||||
res.setHeader(
|
||||
'access-control-allow-headers',
|
||||
'content-type, authorization, x-api-key, x-admin-secret'
|
||||
);
|
||||
}
|
||||
|
||||
function sendJson(res, status, body) {
|
||||
withCors(res);
|
||||
res.statusCode = status;
|
||||
res.setHeader('content-type', 'application/json; charset=utf-8');
|
||||
res.end(JSON.stringify(body));
|
||||
}
|
||||
|
||||
async function readBodyJson(req, { maxBytes }) {
|
||||
const chunks = [];
|
||||
let total = 0;
|
||||
for await (const chunk of req) {
|
||||
total += chunk.length;
|
||||
if (total > maxBytes) throw new Error('payload_too_large');
|
||||
chunks.push(chunk);
|
||||
}
|
||||
const text = Buffer.concat(chunks).toString('utf8');
|
||||
if (!text.trim()) return {};
|
||||
try {
|
||||
return JSON.parse(text);
|
||||
} catch {
|
||||
throw new Error('invalid_json');
|
||||
}
|
||||
}
|
||||
|
||||
function parseNumeric(value) {
|
||||
if (value == null) return null;
|
||||
if (typeof value === 'number') return Number.isFinite(value) ? value : null;
|
||||
if (typeof value === 'string') {
|
||||
const s = value.trim();
|
||||
if (!s) return null;
|
||||
const n = Number(s);
|
||||
return Number.isFinite(n) ? n : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function clampInt(value, min, max) {
|
||||
const n = Number.parseInt(String(value), 10);
|
||||
if (!Number.isFinite(n) || !Number.isInteger(n)) return min;
|
||||
return Math.min(max, Math.max(min, n));
|
||||
}
|
||||
|
||||
function clampNumber(value, min, max, fallback) {
|
||||
const n = typeof value === 'number' ? value : Number(value);
|
||||
if (!Number.isFinite(n)) return fallback;
|
||||
return Math.min(max, Math.max(min, n));
|
||||
}
|
||||
|
||||
function isUuid(value) {
|
||||
const s = String(value || '').trim();
|
||||
return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(s);
|
||||
}
|
||||
|
||||
async function hasuraRequest(query, variables) {
|
||||
if (!HASURA_ADMIN_SECRET) throw new Error('Missing HASURA_ADMIN_SECRET');
|
||||
const res = await fetch(HASURA_URL, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'content-type': 'application/json',
|
||||
'x-hasura-admin-secret': HASURA_ADMIN_SECRET,
|
||||
},
|
||||
body: JSON.stringify({ query, variables }),
|
||||
});
|
||||
const text = await res.text();
|
||||
if (!res.ok) throw new Error(`Hasura HTTP ${res.status}: ${text}`);
|
||||
const json = text ? JSON.parse(text) : {};
|
||||
if (json.errors?.length) throw new Error(json.errors.map((e) => e.message).join(' | '));
|
||||
return json.data;
|
||||
}
|
||||
|
||||
function normalizeScopes(value) {
|
||||
if (!value) return [];
|
||||
if (Array.isArray(value)) return value.map((v) => String(v)).filter(Boolean);
|
||||
if (typeof value === 'string') {
|
||||
return value
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean);
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
async function requireValidToken(req, requiredScope) {
|
||||
const token = readBearerToken(req);
|
||||
if (!token) return { ok: false, status: 401, error: 'missing_token' };
|
||||
|
||||
const hash = sha256Hex(token);
|
||||
const query = `
|
||||
query ValidToken($hash: String!) {
|
||||
api_tokens(where: {token_hash: {_eq: $hash}, revoked_at: {_is_null: true}}, limit: 1) {
|
||||
id
|
||||
name
|
||||
scopes
|
||||
}
|
||||
}
|
||||
`;
|
||||
|
||||
let data;
|
||||
try {
|
||||
data = await hasuraRequest(query, { hash });
|
||||
} catch (err) {
|
||||
return { ok: false, status: 500, error: String(err?.message || err) };
|
||||
}
|
||||
|
||||
const row = data?.api_tokens?.[0];
|
||||
if (!row?.id) return { ok: false, status: 401, error: 'invalid_or_revoked_token' };
|
||||
const scopes = normalizeScopes(row.scopes);
|
||||
if (requiredScope && !scopes.includes(requiredScope)) {
|
||||
return { ok: false, status: 403, error: 'missing_scope' };
|
||||
}
|
||||
|
||||
// best-effort touch
|
||||
const touch = `
|
||||
mutation TouchToken($id: uuid!, $ts: timestamptz!) {
|
||||
update_api_tokens_by_pk(pk_columns: {id: $id}, _set: {last_used_at: $ts}) { id }
|
||||
}
|
||||
`;
|
||||
hasuraRequest(touch, { id: row.id, ts: getIsoNow() }).catch(() => {});
|
||||
|
||||
return { ok: true, token: { id: row.id, name: row.name } };
|
||||
}
|
||||
|
||||
async function readSolPriceUsd() {
|
||||
try {
|
||||
const q = `
|
||||
query SolPrice {
|
||||
dlob_stats_latest(where: {market_name: {_eq: "SOL-PERP"}}, limit: 1) {
|
||||
mid_price
|
||||
mark_price
|
||||
oracle_price
|
||||
}
|
||||
}
|
||||
`;
|
||||
const data = await hasuraRequest(q, {});
|
||||
const row = data?.dlob_stats_latest?.[0];
|
||||
const p = parseNumeric(row?.mid_price) ?? parseNumeric(row?.mark_price) ?? parseNumeric(row?.oracle_price);
|
||||
if (p != null && p > 0) return p;
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function getByPath(obj, pathStr) {
|
||||
if (!obj || typeof obj !== 'object') return undefined;
|
||||
const parts = String(pathStr || '').split('.').filter(Boolean);
|
||||
let cur = obj;
|
||||
for (const p of parts) {
|
||||
if (!cur || typeof cur !== 'object') return undefined;
|
||||
cur = cur[p];
|
||||
}
|
||||
return cur;
|
||||
}
|
||||
|
||||
function readNumberFromPayload(payload, paths) {
|
||||
for (const p of paths) {
|
||||
const v = getByPath(payload, p);
|
||||
const n = parseNumeric(v);
|
||||
if (n != null) return n;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function readTextFromPayload(payload, paths) {
|
||||
for (const p of paths) {
|
||||
const v = getByPath(payload, p);
|
||||
if (typeof v === 'string' && v.trim()) return v.trim();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function inferContractSizeUsd(contract) {
|
||||
return (
|
||||
readNumberFromPayload(contract, [
|
||||
'desired.size_usd',
|
||||
'desired.sizeUsd',
|
||||
'desired.notional_usd',
|
||||
'desired.notionalUsd',
|
||||
'entry.size_usd',
|
||||
'entry.sizeUsd',
|
||||
'entry.notional_usd',
|
||||
'entry.notionalUsd',
|
||||
'entry.order_intent.size_usd',
|
||||
'entry.order_intent.sizeUsd',
|
||||
'desired.order_intent.size_usd',
|
||||
'desired.order_intent.sizeUsd',
|
||||
]) || null
|
||||
);
|
||||
}
|
||||
|
||||
function inferContractSide(contract) {
|
||||
const raw =
|
||||
readTextFromPayload(contract, [
|
||||
'desired.side',
|
||||
'entry.side',
|
||||
'entry.order_intent.side',
|
||||
'desired.order_intent.side',
|
||||
'desired.direction',
|
||||
'entry.direction',
|
||||
]) || '';
|
||||
const v = raw.toLowerCase();
|
||||
if (v === 'long' || v === 'buy') return 'long';
|
||||
if (v === 'short' || v === 'sell') return 'short';
|
||||
return null;
|
||||
}
|
||||
|
||||
function sumCostsFromEvents(events) {
|
||||
const totals = {
|
||||
tradeFeeUsd: 0,
|
||||
txFeeUsd: 0,
|
||||
slippageUsd: 0,
|
||||
fundingUsd: 0,
|
||||
realizedPnlUsd: 0,
|
||||
txCount: 0,
|
||||
fillCount: 0,
|
||||
cancelCount: 0,
|
||||
modifyCount: 0,
|
||||
errorCount: 0,
|
||||
};
|
||||
|
||||
for (const ev of events || []) {
|
||||
const t = String(ev?.event_type || '').toLowerCase();
|
||||
const payload = ev?.payload && typeof ev.payload === 'object' ? ev.payload : {};
|
||||
|
||||
const tradeFeeUsd =
|
||||
readNumberFromPayload(payload, [
|
||||
'realized_fee_usd',
|
||||
'trade_fee_usd',
|
||||
'fee_usd',
|
||||
'fees.trade_fee_usd',
|
||||
'fees.usd',
|
||||
]) || 0;
|
||||
const txFeeUsd =
|
||||
readNumberFromPayload(payload, [
|
||||
'realized_tx_usd',
|
||||
'tx_fee_usd',
|
||||
'network_fee_usd',
|
||||
'fees.tx_fee_usd',
|
||||
'fees.network_usd',
|
||||
]) || 0;
|
||||
const slippageUsd =
|
||||
readNumberFromPayload(payload, [
|
||||
'slippage_usd',
|
||||
'realized_slippage_usd',
|
||||
'execution_usd',
|
||||
'realized_execution_usd',
|
||||
]) || 0;
|
||||
const fundingUsd = readNumberFromPayload(payload, ['funding_usd', 'realized_funding_usd']) || 0;
|
||||
const pnlUsd = readNumberFromPayload(payload, ['realized_pnl_usd', 'pnl_usd']) || 0;
|
||||
const txCount = readNumberFromPayload(payload, ['tx_count', 'txCount']) || 0;
|
||||
|
||||
totals.tradeFeeUsd += tradeFeeUsd;
|
||||
totals.txFeeUsd += txFeeUsd;
|
||||
totals.slippageUsd += slippageUsd;
|
||||
totals.fundingUsd += fundingUsd;
|
||||
totals.realizedPnlUsd += pnlUsd;
|
||||
totals.txCount += txCount;
|
||||
|
||||
if (t.includes('fill')) totals.fillCount += 1;
|
||||
if (t.includes('cancel')) totals.cancelCount += 1;
|
||||
if (t.includes('modify') || t.includes('reprice')) totals.modifyCount += 1;
|
||||
if (t.includes('error') || String(ev?.severity || '').toLowerCase() === 'error') totals.errorCount += 1;
|
||||
}
|
||||
|
||||
const totalCostsUsd = totals.tradeFeeUsd + totals.txFeeUsd + totals.slippageUsd + totals.fundingUsd;
|
||||
|
||||
return {
|
||||
...totals,
|
||||
totalCostsUsd,
|
||||
netPnlUsd: totals.realizedPnlUsd - totalCostsUsd,
|
||||
};
|
||||
}
|
||||
|
||||
function buildCostSeriesFromEvents(events, { maxPoints }) {
|
||||
const points = [];
|
||||
const totals = {
|
||||
tradeFeeUsd: 0,
|
||||
txFeeUsd: 0,
|
||||
slippageUsd: 0,
|
||||
fundingUsd: 0,
|
||||
realizedPnlUsd: 0,
|
||||
};
|
||||
|
||||
for (const ev of events || []) {
|
||||
const ts = ev?.ts;
|
||||
if (!ts) continue;
|
||||
const payload = ev?.payload && typeof ev.payload === 'object' ? ev.payload : {};
|
||||
|
||||
const tradeFeeUsd =
|
||||
readNumberFromPayload(payload, [
|
||||
'realized_fee_usd',
|
||||
'trade_fee_usd',
|
||||
'fee_usd',
|
||||
'fees.trade_fee_usd',
|
||||
'fees.usd',
|
||||
]) || 0;
|
||||
const txFeeUsd =
|
||||
readNumberFromPayload(payload, [
|
||||
'realized_tx_usd',
|
||||
'tx_fee_usd',
|
||||
'network_fee_usd',
|
||||
'fees.tx_fee_usd',
|
||||
'fees.network_usd',
|
||||
]) || 0;
|
||||
const slippageUsd =
|
||||
readNumberFromPayload(payload, [
|
||||
'slippage_usd',
|
||||
'realized_slippage_usd',
|
||||
'execution_usd',
|
||||
'realized_execution_usd',
|
||||
]) || 0;
|
||||
const fundingUsd = readNumberFromPayload(payload, ['funding_usd', 'realized_funding_usd']) || 0;
|
||||
const pnlUsd = readNumberFromPayload(payload, ['realized_pnl_usd', 'pnl_usd']) || 0;
|
||||
|
||||
totals.tradeFeeUsd += tradeFeeUsd;
|
||||
totals.txFeeUsd += txFeeUsd;
|
||||
totals.slippageUsd += slippageUsd;
|
||||
totals.fundingUsd += fundingUsd;
|
||||
totals.realizedPnlUsd += pnlUsd;
|
||||
|
||||
const totalCostsUsd = totals.tradeFeeUsd + totals.txFeeUsd + totals.slippageUsd + totals.fundingUsd;
|
||||
points.push({
|
||||
ts,
|
||||
tradeFeeUsd: totals.tradeFeeUsd,
|
||||
txFeeUsd: totals.txFeeUsd,
|
||||
slippageUsd: totals.slippageUsd,
|
||||
fundingUsd: totals.fundingUsd,
|
||||
totalCostsUsd,
|
||||
realizedPnlUsd: totals.realizedPnlUsd,
|
||||
netPnlUsd: totals.realizedPnlUsd - totalCostsUsd,
|
||||
});
|
||||
}
|
||||
|
||||
const cap = Math.max(50, Math.min(10_000, Number(maxPoints) || 600));
|
||||
if (points.length <= cap) return points;
|
||||
|
||||
const step = Math.ceil(points.length / cap);
|
||||
const sampled = [];
|
||||
for (let i = 0; i < points.length; i += step) sampled.push(points[i]);
|
||||
const last = points[points.length - 1];
|
||||
if (sampled[sampled.length - 1] !== last) sampled.push(last);
|
||||
return sampled;
|
||||
}
|
||||
|
||||
function proxyToUpstream(req, res) {
|
||||
const url = new URL(req.url || '/', `http://${req.headers.host || 'localhost'}`);
|
||||
|
||||
const headers = { ...req.headers };
|
||||
delete headers.host;
|
||||
delete headers.connection;
|
||||
|
||||
const opts = {
|
||||
host: UPSTREAM_HOST,
|
||||
port: UPSTREAM_PORT,
|
||||
method: req.method,
|
||||
path: url.pathname + url.search,
|
||||
headers,
|
||||
};
|
||||
|
||||
const upstreamReq = http.request(opts, (upstreamRes) => {
|
||||
withCors(res);
|
||||
res.statusCode = upstreamRes.statusCode || 502;
|
||||
|
||||
for (const [k, v] of Object.entries(upstreamRes.headers || {})) {
|
||||
if (!k) continue;
|
||||
if (k.toLowerCase() === 'content-length') continue;
|
||||
if (k.toLowerCase().startsWith('access-control-')) continue;
|
||||
if (v != null) res.setHeader(k, v);
|
||||
}
|
||||
|
||||
upstreamRes.pipe(res);
|
||||
});
|
||||
|
||||
upstreamReq.on('error', (err) => {
|
||||
sendJson(res, 502, { ok: false, error: String(err?.message || err) });
|
||||
});
|
||||
|
||||
req.pipe(upstreamReq);
|
||||
}
|
||||
|
||||
async function handleMonitor(req, res, url) {
|
||||
const auth = await requireValidToken(req, 'read');
|
||||
if (!auth.ok) {
|
||||
sendJson(res, auth.status, { ok: false, error: auth.error });
|
||||
return;
|
||||
}
|
||||
|
||||
const pathname = url.pathname;
|
||||
const parts = pathname.split('/').filter(Boolean);
|
||||
const contractId = parts[2];
|
||||
if (!isUuid(contractId)) {
|
||||
sendJson(res, 400, { ok: false, error: 'invalid_contract_id' });
|
||||
return;
|
||||
}
|
||||
|
||||
const limit = clampInt(url.searchParams.get('eventsLimit') || '2000', 10, 50_000);
|
||||
const wantSeries = (url.searchParams.get('series') || '').trim() === '1';
|
||||
const seriesMax = clampInt(url.searchParams.get('seriesMax') || '600', 50, 10_000);
|
||||
|
||||
const qContract = `
|
||||
query ContractByPk($id: uuid!) {
|
||||
bot_contracts_by_pk(id: $id) {
|
||||
id
|
||||
decision_id
|
||||
bot_id
|
||||
model_version
|
||||
market_name
|
||||
subaccount_id
|
||||
status
|
||||
desired
|
||||
entry
|
||||
manage
|
||||
exit
|
||||
gates
|
||||
created_at
|
||||
updated_at
|
||||
last_heartbeat_at
|
||||
ended_at
|
||||
reason
|
||||
}
|
||||
}
|
||||
`;
|
||||
const qEvents = `
|
||||
query ContractEvents($id: uuid!, $limit: Int!) {
|
||||
bot_events(where: {contract_id: {_eq: $id}}, order_by: {ts: asc}, limit: $limit) {
|
||||
ts
|
||||
contract_id
|
||||
decision_id
|
||||
bot_id
|
||||
market_name
|
||||
event_type
|
||||
severity
|
||||
payload
|
||||
}
|
||||
}
|
||||
`;
|
||||
|
||||
try {
|
||||
const data = await hasuraRequest(qContract, { id: contractId });
|
||||
const contract = data?.bot_contracts_by_pk;
|
||||
if (!contract?.id) {
|
||||
sendJson(res, 404, { ok: false, error: 'contract_not_found' });
|
||||
return;
|
||||
}
|
||||
|
||||
const evData = await hasuraRequest(qEvents, { id: contractId, limit });
|
||||
const events = evData?.bot_events || [];
|
||||
const costs = sumCostsFromEvents(events);
|
||||
const series = wantSeries ? buildCostSeriesFromEvents(events, { maxPoints: seriesMax }) : null;
|
||||
|
||||
const sizeUsd = inferContractSizeUsd(contract);
|
||||
const side = inferContractSide(contract);
|
||||
|
||||
let closeEst = null;
|
||||
if (contract.market_name && sizeUsd != null) {
|
||||
const qSlip = `
|
||||
query Slippage($market: String!) {
|
||||
dlob_slippage_latest(where: {market_name: {_eq: $market}}) {
|
||||
market_name
|
||||
side
|
||||
size_usd
|
||||
mid_price
|
||||
vwap_price
|
||||
worst_price
|
||||
impact_bps
|
||||
fill_pct
|
||||
updated_at
|
||||
}
|
||||
}
|
||||
`;
|
||||
const slipData = await hasuraRequest(qSlip, { market: contract.market_name });
|
||||
const rows = slipData?.dlob_slippage_latest || [];
|
||||
|
||||
const pickNearest = (wantedSide) => {
|
||||
const candidates = rows.filter((r) => String(r.side || '').toLowerCase() === wantedSide);
|
||||
if (!candidates.length) return null;
|
||||
let best = null;
|
||||
let bestD = Infinity;
|
||||
for (const r of candidates) {
|
||||
const s = parseNumeric(r.size_usd);
|
||||
if (s == null) continue;
|
||||
const d = Math.abs(s - sizeUsd);
|
||||
if (d < bestD) {
|
||||
bestD = d;
|
||||
best = r;
|
||||
}
|
||||
}
|
||||
return best;
|
||||
};
|
||||
|
||||
closeEst = {
|
||||
requestedSizeUsd: sizeUsd,
|
||||
entrySide: side,
|
||||
suggestedCloseSide: side === 'long' ? 'sell' : side === 'short' ? 'buy' : null,
|
||||
buy: pickNearest('buy'),
|
||||
sell: pickNearest('sell'),
|
||||
};
|
||||
}
|
||||
|
||||
sendJson(res, 200, {
|
||||
ok: true,
|
||||
contract,
|
||||
eventsCount: events.length,
|
||||
costs,
|
||||
series,
|
||||
closeEstimate: closeEst,
|
||||
});
|
||||
} catch (err) {
|
||||
sendJson(res, 500, { ok: false, error: String(err?.message || err) });
|
||||
}
|
||||
}
|
||||
|
||||
async function handleEstimate(req, res) {
|
||||
const auth = await requireValidToken(req, 'read');
|
||||
if (!auth.ok) {
|
||||
sendJson(res, auth.status, { ok: false, error: auth.error });
|
||||
return;
|
||||
}
|
||||
|
||||
let body;
|
||||
try {
|
||||
body = await readBodyJson(req, { maxBytes: 256 * 1024 });
|
||||
} catch (err) {
|
||||
const msg = String(err?.message || err);
|
||||
if (msg === 'payload_too_large') {
|
||||
sendJson(res, 413, { ok: false, error: 'payload_too_large' });
|
||||
return;
|
||||
}
|
||||
sendJson(res, 400, { ok: false, error: 'invalid_json' });
|
||||
return;
|
||||
}
|
||||
|
||||
const market = String(body?.market_name || body?.market || '').trim();
|
||||
if (!market) {
|
||||
sendJson(res, 400, { ok: false, error: 'missing_market' });
|
||||
return;
|
||||
}
|
||||
|
||||
const notionalUsd = parseNumeric(body?.notional_usd ?? body?.notionalUsd ?? body?.size_usd ?? body?.sizeUsd);
|
||||
if (!(notionalUsd != null && notionalUsd > 0)) {
|
||||
sendJson(res, 400, { ok: false, error: 'invalid_notional_usd' });
|
||||
return;
|
||||
}
|
||||
|
||||
const entrySideRaw = String(body?.side || 'long').trim().toLowerCase();
|
||||
const entrySide = entrySideRaw === 'short' ? 'short' : 'long';
|
||||
|
||||
const orderType = String(body?.order_type || body?.orderType || 'market').trim().toLowerCase();
|
||||
const isMarket = orderType === 'market' || orderType === 'taker';
|
||||
|
||||
const takerBps = clampNumber(parseNumeric(body?.fee_taker_bps) ?? 5, 0, 1000, 5);
|
||||
const makerBps = clampNumber(parseNumeric(body?.fee_maker_bps) ?? 0, -1000, 1000, 0);
|
||||
const feeBps = isMarket ? takerBps : makerBps;
|
||||
|
||||
let txFeeUsdEst = parseNumeric(body?.tx_fee_usd_est);
|
||||
if (txFeeUsdEst == null) {
|
||||
const baseLamports = 5000;
|
||||
const sigs = 1;
|
||||
const priorityLamports = 0;
|
||||
const lamports = baseLamports * sigs + priorityLamports;
|
||||
const sol = lamports / 1_000_000_000;
|
||||
const solUsd = await readSolPriceUsd();
|
||||
txFeeUsdEst = solUsd != null ? sol * solUsd : 0;
|
||||
}
|
||||
txFeeUsdEst = clampNumber(txFeeUsdEst, 0, 100, 0);
|
||||
|
||||
const expectedReprices = clampInt(body?.expected_reprices_per_entry ?? body?.expectedReprices ?? '0', 0, 500);
|
||||
const modifyTxCount = clampInt(body?.modify_tx_count ?? body?.modifyTxCount ?? '2', 0, 10);
|
||||
|
||||
try {
|
||||
const qSlip = `
|
||||
query Slippage($market: String!) {
|
||||
dlob_slippage_latest(where: {market_name: {_eq: $market}}) {
|
||||
market_name
|
||||
side
|
||||
size_usd
|
||||
mid_price
|
||||
vwap_price
|
||||
worst_price
|
||||
impact_bps
|
||||
fill_pct
|
||||
updated_at
|
||||
}
|
||||
}
|
||||
`;
|
||||
const slipData = await hasuraRequest(qSlip, { market });
|
||||
const rows = slipData?.dlob_slippage_latest || [];
|
||||
const wantedSide = entrySide === 'long' ? 'buy' : 'sell';
|
||||
const candidates = rows.filter((r) => String(r.side || '').toLowerCase() === wantedSide);
|
||||
|
||||
let best = null;
|
||||
let bestD = Infinity;
|
||||
for (const r of candidates) {
|
||||
const s = parseNumeric(r.size_usd);
|
||||
if (s == null) continue;
|
||||
const d = Math.abs(s - notionalUsd);
|
||||
if (d < bestD) {
|
||||
bestD = d;
|
||||
best = r;
|
||||
}
|
||||
}
|
||||
|
||||
const impactBps = parseNumeric(best?.impact_bps) ?? 0;
|
||||
const slippageUsd = (notionalUsd * impactBps) / 10_000;
|
||||
const tradeFeeUsd = (notionalUsd * feeBps) / 10_000;
|
||||
const modifyCostUsd = expectedReprices * modifyTxCount * txFeeUsdEst;
|
||||
const totalUsd = tradeFeeUsd + slippageUsd + txFeeUsdEst + modifyCostUsd;
|
||||
const totalBps = (totalUsd / notionalUsd) * 10_000;
|
||||
|
||||
sendJson(res, 200, {
|
||||
ok: true,
|
||||
input: {
|
||||
market_name: market,
|
||||
notional_usd: notionalUsd,
|
||||
side: entrySide,
|
||||
order_type: orderType,
|
||||
fee_bps: feeBps,
|
||||
tx_fee_usd_est: txFeeUsdEst,
|
||||
expected_reprices_per_entry: expectedReprices,
|
||||
},
|
||||
dlob: best
|
||||
? {
|
||||
size_usd: best.size_usd,
|
||||
side: best.side,
|
||||
mid_price: best.mid_price,
|
||||
vwap_price: best.vwap_price,
|
||||
impact_bps: best.impact_bps,
|
||||
fill_pct: best.fill_pct,
|
||||
updated_at: best.updated_at,
|
||||
}
|
||||
: null,
|
||||
breakdown: {
|
||||
trade_fee_usd: tradeFeeUsd,
|
||||
slippage_usd: slippageUsd,
|
||||
tx_fee_usd: txFeeUsdEst,
|
||||
expected_modify_usd: modifyCostUsd,
|
||||
total_usd: totalUsd,
|
||||
total_bps: totalBps,
|
||||
breakeven_bps: totalBps,
|
||||
},
|
||||
});
|
||||
} catch (err) {
|
||||
sendJson(res, 500, { ok: false, error: String(err?.message || err) });
|
||||
}
|
||||
}
|
||||
|
||||
let upstreamChild = null;
|
||||
|
||||
function startUpstream() {
|
||||
const env = { ...process.env, PORT: String(UPSTREAM_PORT) };
|
||||
upstreamChild = spawn('node', [UPSTREAM_ENTRY], { env, stdio: 'inherit' });
|
||||
upstreamChild.on('exit', (code, signal) => {
|
||||
console.error(`upstream exited: code=${code} signal=${signal}`);
|
||||
});
|
||||
}
|
||||
|
||||
function shutdown() {
|
||||
if (upstreamChild && !upstreamChild.killed) upstreamChild.kill('SIGTERM');
|
||||
}
|
||||
|
||||
process.on('SIGTERM', shutdown);
|
||||
process.on('SIGINT', shutdown);
|
||||
|
||||
startUpstream();
|
||||
|
||||
const server = http.createServer(async (req, res) => {
|
||||
if (req.method === 'OPTIONS') {
|
||||
withCors(res);
|
||||
res.statusCode = 204;
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
|
||||
const url = new URL(req.url || '/', `http://${req.headers.host || 'localhost'}`);
|
||||
const pathname = url.pathname;
|
||||
|
||||
if (req.method === 'GET' && pathname === '/healthz') {
|
||||
// Check upstream quickly; if it's down, we fail readiness.
|
||||
const opts = { host: UPSTREAM_HOST, port: UPSTREAM_PORT, path: '/healthz', method: 'GET', timeout: 800 };
|
||||
const upstreamOk = await new Promise((resolve) => {
|
||||
const r = http.request(opts, (rr) => {
|
||||
rr.resume();
|
||||
resolve(rr.statusCode === 200);
|
||||
});
|
||||
r.on('timeout', () => {
|
||||
r.destroy();
|
||||
resolve(false);
|
||||
});
|
||||
r.on('error', () => resolve(false));
|
||||
r.end();
|
||||
});
|
||||
|
||||
if (!upstreamOk) {
|
||||
sendJson(res, 503, { ok: false, error: 'upstream_not_ready' });
|
||||
return;
|
||||
}
|
||||
|
||||
sendJson(res, 200, {
|
||||
ok: true,
|
||||
service: 'trade-api-wrapper',
|
||||
startedAt: getIsoNow(),
|
||||
upstream: { host: UPSTREAM_HOST, port: UPSTREAM_PORT },
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (req.method === 'POST' && pathname === '/v1/contracts/costs/estimate') {
|
||||
await handleEstimate(req, res);
|
||||
return;
|
||||
}
|
||||
|
||||
if (req.method === 'GET' && pathname.startsWith('/v1/contracts/') && pathname.endsWith('/monitor')) {
|
||||
await handleMonitor(req, res, url);
|
||||
return;
|
||||
}
|
||||
|
||||
proxyToUpstream(req, res);
|
||||
});
|
||||
|
||||
server.listen(WRAPPER_PORT, () => {
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
service: 'trade-api-wrapper',
|
||||
port: WRAPPER_PORT,
|
||||
upstream: { entry: UPSTREAM_ENTRY, host: UPSTREAM_HOST, port: UPSTREAM_PORT },
|
||||
hasuraUrl: HASURA_URL,
|
||||
hasuraAdminSecret: HASURA_ADMIN_SECRET ? '***' : undefined,
|
||||
},
|
||||
null,
|
||||
2
|
||||
)
|
||||
);
|
||||
});
|
||||
@@ -1,50 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-depth-worker-drift
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "6"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-depth-worker-drift
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-depth-worker-drift
|
||||
spec:
|
||||
containers:
|
||||
- name: worker
|
||||
image: node:20-slim
|
||||
imagePullPolicy: IfNotPresent
|
||||
env:
|
||||
- name: HASURA_GRAPHQL_URL
|
||||
value: http://hasura:8080/v1/graphql
|
||||
- name: HASURA_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-hasura
|
||||
key: HASURA_GRAPHQL_ADMIN_SECRET
|
||||
- name: DLOB_SOURCE
|
||||
value: drift
|
||||
- name: DLOB_MARKETS
|
||||
value: SOL-PERP,DOGE-PERP,JUP-PERP
|
||||
- name: DLOB_POLL_MS
|
||||
value: "1000"
|
||||
- name: DLOB_DEPTH_BPS_BANDS
|
||||
value: "5,10,20,50,100,200"
|
||||
- name: PRICE_PRECISION
|
||||
value: "1000000"
|
||||
- name: BASE_PRECISION
|
||||
value: "1000000000"
|
||||
command: ["node", "/app/worker.mjs"]
|
||||
volumeMounts:
|
||||
- name: script
|
||||
mountPath: /app/worker.mjs
|
||||
subPath: worker.mjs
|
||||
readOnly: true
|
||||
volumes:
|
||||
- name: script
|
||||
configMap:
|
||||
name: dlob-depth-worker-script
|
||||
@@ -1,50 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-depth-worker
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "6"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-depth-worker
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-depth-worker
|
||||
spec:
|
||||
containers:
|
||||
- name: worker
|
||||
image: node:20-slim
|
||||
imagePullPolicy: IfNotPresent
|
||||
env:
|
||||
- name: HASURA_GRAPHQL_URL
|
||||
value: http://hasura:8080/v1/graphql
|
||||
- name: HASURA_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-hasura
|
||||
key: HASURA_GRAPHQL_ADMIN_SECRET
|
||||
- name: DLOB_SOURCE
|
||||
value: mevnode
|
||||
- name: DLOB_MARKETS
|
||||
value: SOL-PERP,DOGE-PERP,JUP-PERP
|
||||
- name: DLOB_POLL_MS
|
||||
value: "1000"
|
||||
- name: DLOB_DEPTH_BPS_BANDS
|
||||
value: "5,10,20,50,100,200"
|
||||
- name: PRICE_PRECISION
|
||||
value: "1000000"
|
||||
- name: BASE_PRECISION
|
||||
value: "1000000000"
|
||||
command: ["node", "/app/worker.mjs"]
|
||||
volumeMounts:
|
||||
- name: script
|
||||
mountPath: /app/worker.mjs
|
||||
subPath: worker.mjs
|
||||
readOnly: true
|
||||
volumes:
|
||||
- name: script
|
||||
configMap:
|
||||
name: dlob-depth-worker-script
|
||||
@@ -1,316 +0,0 @@
|
||||
import process from 'node:process';
|
||||
import { setTimeout as sleep } from 'node:timers/promises';
|
||||
|
||||
function getIsoNow() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function clampInt(value, min, max, fallback) {
|
||||
const n = Number.parseInt(String(value ?? ''), 10);
|
||||
if (!Number.isInteger(n)) return fallback;
|
||||
return Math.min(max, Math.max(min, n));
|
||||
}
|
||||
|
||||
function envList(name, fallbackCsv) {
|
||||
const raw = process.env[name] ?? fallbackCsv;
|
||||
return String(raw)
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
function envIntList(name, fallbackCsv) {
|
||||
const out = [];
|
||||
for (const item of envList(name, fallbackCsv)) {
|
||||
const n = Number.parseInt(item, 10);
|
||||
if (!Number.isFinite(n)) continue;
|
||||
out.push(n);
|
||||
}
|
||||
return out.length ? out : envList(name, fallbackCsv).map((v) => Number.parseInt(v, 10)).filter(Number.isFinite);
|
||||
}
|
||||
|
||||
function toNumberOrNull(value) {
|
||||
if (value == null) return null;
|
||||
if (typeof value === 'number') return Number.isFinite(value) ? value : null;
|
||||
if (typeof value === 'string') {
|
||||
const s = value.trim();
|
||||
if (!s) return null;
|
||||
const n = Number(s);
|
||||
return Number.isFinite(n) ? n : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function numStr(value) {
|
||||
if (value == null) return null;
|
||||
if (typeof value === 'number') return Number.isFinite(value) ? String(value) : null;
|
||||
if (typeof value === 'string') return value.trim() || null;
|
||||
return null;
|
||||
}
|
||||
|
||||
function jsonNormalize(value) {
|
||||
if (typeof value !== 'string') return value;
|
||||
const s = value.trim();
|
||||
if (!s) return null;
|
||||
try {
|
||||
return JSON.parse(s);
|
||||
} catch {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
function resolveConfig() {
|
||||
const hasuraUrl = process.env.HASURA_GRAPHQL_URL || 'http://hasura:8080/v1/graphql';
|
||||
const hasuraAdminSecret = process.env.HASURA_ADMIN_SECRET || process.env.HASURA_GRAPHQL_ADMIN_SECRET || undefined;
|
||||
const hasuraAuthToken = process.env.HASURA_AUTH_TOKEN || process.env.HASURA_JWT || undefined;
|
||||
|
||||
const dlobSource = String(process.env.DLOB_SOURCE || 'mevnode').trim() || 'mevnode';
|
||||
const markets = envList('DLOB_MARKETS', 'SOL-PERP,DOGE-PERP,JUP-PERP');
|
||||
const pollMs = clampInt(process.env.DLOB_POLL_MS, 250, 60_000, 1000);
|
||||
const bandsBps = envIntList('DLOB_DEPTH_BPS_BANDS', '5,10,20,50,100,200');
|
||||
|
||||
const pricePrecision = Number(process.env.PRICE_PRECISION || 1_000_000);
|
||||
const basePrecision = Number(process.env.BASE_PRECISION || 1_000_000_000);
|
||||
if (!Number.isFinite(pricePrecision) || pricePrecision <= 0)
|
||||
throw new Error(`Invalid PRICE_PRECISION: ${process.env.PRICE_PRECISION}`);
|
||||
if (!Number.isFinite(basePrecision) || basePrecision <= 0)
|
||||
throw new Error(`Invalid BASE_PRECISION: ${process.env.BASE_PRECISION}`);
|
||||
|
||||
return {
|
||||
hasuraUrl,
|
||||
hasuraAdminSecret,
|
||||
hasuraAuthToken,
|
||||
dlobSource,
|
||||
markets,
|
||||
pollMs,
|
||||
bandsBps,
|
||||
pricePrecision,
|
||||
basePrecision,
|
||||
};
|
||||
}
|
||||
|
||||
async function graphqlRequest(cfg, query, variables) {
|
||||
const headers = { 'content-type': 'application/json' };
|
||||
if (cfg.hasuraAuthToken) {
|
||||
headers.authorization = `Bearer ${cfg.hasuraAuthToken}`;
|
||||
} else if (cfg.hasuraAdminSecret) {
|
||||
headers['x-hasura-admin-secret'] = cfg.hasuraAdminSecret;
|
||||
} else {
|
||||
throw new Error('Missing Hasura auth (set HASURA_AUTH_TOKEN or HASURA_ADMIN_SECRET)');
|
||||
}
|
||||
|
||||
const res = await fetch(cfg.hasuraUrl, {
|
||||
method: 'POST',
|
||||
headers,
|
||||
body: JSON.stringify({ query, variables }),
|
||||
signal: AbortSignal.timeout(10_000),
|
||||
});
|
||||
const text = await res.text();
|
||||
if (!res.ok) throw new Error(`Hasura HTTP ${res.status}: ${text}`);
|
||||
const json = JSON.parse(text);
|
||||
if (json.errors?.length) throw new Error(json.errors.map((e) => e.message).join(' | '));
|
||||
return json.data;
|
||||
}
|
||||
|
||||
function parseLevels(raw, pricePrecision, basePrecision, side) {
|
||||
const v = jsonNormalize(raw);
|
||||
if (!Array.isArray(v)) return [];
|
||||
|
||||
const out = [];
|
||||
for (const item of v) {
|
||||
const priceInt = toNumberOrNull(item?.price);
|
||||
const sizeInt = toNumberOrNull(item?.size);
|
||||
if (priceInt == null || sizeInt == null) continue;
|
||||
const price = priceInt / pricePrecision;
|
||||
const size = sizeInt / basePrecision;
|
||||
if (!Number.isFinite(price) || !Number.isFinite(size)) continue;
|
||||
out.push({ price, size });
|
||||
}
|
||||
|
||||
if (side === 'bid') out.sort((a, b) => b.price - a.price);
|
||||
if (side === 'ask') out.sort((a, b) => a.price - b.price);
|
||||
return out;
|
||||
}
|
||||
|
||||
function computeMid(bestBid, bestAsk, markPrice, oraclePrice) {
|
||||
if (bestBid != null && bestAsk != null) return (bestBid + bestAsk) / 2;
|
||||
if (markPrice != null) return markPrice;
|
||||
if (oraclePrice != null) return oraclePrice;
|
||||
return null;
|
||||
}
|
||||
|
||||
function computeBandDepth({ bids, asks, mid, bandBps }) {
|
||||
if (mid == null || !(mid > 0)) {
|
||||
return { bidBase: 0, askBase: 0, bidUsd: 0, askUsd: 0, imbalance: null };
|
||||
}
|
||||
|
||||
const minBidPrice = mid * (1 - bandBps / 10_000);
|
||||
const maxAskPrice = mid * (1 + bandBps / 10_000);
|
||||
|
||||
let bidBase = 0;
|
||||
let askBase = 0;
|
||||
let bidUsd = 0;
|
||||
let askUsd = 0;
|
||||
|
||||
for (const lvl of bids) {
|
||||
if (lvl.price < minBidPrice) break;
|
||||
bidBase += lvl.size;
|
||||
bidUsd += lvl.size * lvl.price;
|
||||
}
|
||||
|
||||
for (const lvl of asks) {
|
||||
if (lvl.price > maxAskPrice) break;
|
||||
askBase += lvl.size;
|
||||
askUsd += lvl.size * lvl.price;
|
||||
}
|
||||
|
||||
const denom = bidUsd + askUsd;
|
||||
const imbalance = denom > 0 ? (bidUsd - askUsd) / denom : null;
|
||||
return { bidBase, askBase, bidUsd, askUsd, imbalance };
|
||||
}
|
||||
|
||||
async function fetchL2Latest(cfg) {
|
||||
const query = `
|
||||
query DlobL2Latest($source: String!, $markets: [String!]!) {
|
||||
dlob_l2_latest(where: {source: {_eq: $source}, market_name: {_in: $markets}}) {
|
||||
source
|
||||
market_name
|
||||
market_type
|
||||
market_index
|
||||
ts
|
||||
slot
|
||||
mark_price
|
||||
oracle_price
|
||||
best_bid_price
|
||||
best_ask_price
|
||||
bids
|
||||
asks
|
||||
updated_at
|
||||
}
|
||||
}
|
||||
`;
|
||||
const data = await graphqlRequest(cfg, query, { source: cfg.dlobSource, markets: cfg.markets });
|
||||
return Array.isArray(data?.dlob_l2_latest) ? data.dlob_l2_latest : [];
|
||||
}
|
||||
|
||||
async function upsertDepth(cfg, rows) {
|
||||
if (!rows.length) return;
|
||||
const mutation = `
|
||||
mutation UpsertDlobDepth($rows: [dlob_depth_bps_latest_insert_input!]!) {
|
||||
insert_dlob_depth_bps_latest(
|
||||
objects: $rows
|
||||
on_conflict: {
|
||||
constraint: dlob_depth_bps_latest_pkey
|
||||
update_columns: [
|
||||
market_type
|
||||
market_index
|
||||
ts
|
||||
slot
|
||||
mid_price
|
||||
best_bid_price
|
||||
best_ask_price
|
||||
bid_base
|
||||
ask_base
|
||||
bid_usd
|
||||
ask_usd
|
||||
imbalance
|
||||
raw
|
||||
updated_at
|
||||
]
|
||||
}
|
||||
) { affected_rows }
|
||||
}
|
||||
`;
|
||||
await graphqlRequest(cfg, mutation, { rows });
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const cfg = resolveConfig();
|
||||
const lastUpdatedAtByMarket = new Map();
|
||||
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
service: 'dlob-depth-worker',
|
||||
startedAt: getIsoNow(),
|
||||
hasuraUrl: cfg.hasuraUrl,
|
||||
hasuraAuth: cfg.hasuraAuthToken ? 'bearer' : cfg.hasuraAdminSecret ? 'admin-secret' : 'none',
|
||||
dlobSource: cfg.dlobSource,
|
||||
markets: cfg.markets,
|
||||
pollMs: cfg.pollMs,
|
||||
bandsBps: cfg.bandsBps,
|
||||
pricePrecision: cfg.pricePrecision,
|
||||
basePrecision: cfg.basePrecision,
|
||||
},
|
||||
null,
|
||||
2
|
||||
)
|
||||
);
|
||||
|
||||
while (true) {
|
||||
const rows = [];
|
||||
|
||||
try {
|
||||
const l2Rows = await fetchL2Latest(cfg);
|
||||
for (const l2 of l2Rows) {
|
||||
const market = String(l2.market_name || '').trim();
|
||||
if (!market) continue;
|
||||
|
||||
const updatedAt = l2.updated_at || null;
|
||||
if (updatedAt && lastUpdatedAtByMarket.get(market) === updatedAt) continue;
|
||||
if (updatedAt) lastUpdatedAtByMarket.set(market, updatedAt);
|
||||
|
||||
const bestBid = toNumberOrNull(l2.best_bid_price);
|
||||
const bestAsk = toNumberOrNull(l2.best_ask_price);
|
||||
const markPrice = toNumberOrNull(l2.mark_price);
|
||||
const oraclePrice = toNumberOrNull(l2.oracle_price);
|
||||
const mid = computeMid(bestBid, bestAsk, markPrice, oraclePrice);
|
||||
|
||||
const bids = parseLevels(l2.bids, cfg.pricePrecision, cfg.basePrecision, 'bid');
|
||||
const asks = parseLevels(l2.asks, cfg.pricePrecision, cfg.basePrecision, 'ask');
|
||||
|
||||
for (const bandBps of cfg.bandsBps) {
|
||||
const d = computeBandDepth({ bids, asks, mid, bandBps });
|
||||
rows.push({
|
||||
source: cfg.dlobSource,
|
||||
market_name: market,
|
||||
band_bps: bandBps,
|
||||
market_type: l2.market_type ? String(l2.market_type) : 'perp',
|
||||
market_index: typeof l2.market_index === 'number' ? l2.market_index : null,
|
||||
ts: l2.ts == null ? null : String(l2.ts),
|
||||
slot: l2.slot == null ? null : String(l2.slot),
|
||||
mid_price: numStr(mid),
|
||||
best_bid_price: numStr(bestBid),
|
||||
best_ask_price: numStr(bestAsk),
|
||||
bid_base: numStr(d.bidBase),
|
||||
ask_base: numStr(d.askBase),
|
||||
bid_usd: numStr(d.bidUsd),
|
||||
ask_usd: numStr(d.askUsd),
|
||||
imbalance: numStr(d.imbalance),
|
||||
raw: {
|
||||
ref: 'mid',
|
||||
pricePrecision: cfg.pricePrecision,
|
||||
basePrecision: cfg.basePrecision,
|
||||
},
|
||||
updated_at: updatedAt,
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`[dlob-depth-worker] fetch/compute: ${String(err?.message || err)}`);
|
||||
}
|
||||
|
||||
try {
|
||||
await upsertDepth(cfg, rows);
|
||||
} catch (err) {
|
||||
console.error(`[dlob-depth-worker] upsert: ${String(err?.message || err)}`);
|
||||
}
|
||||
|
||||
await sleep(cfg.pollMs);
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(String(err?.stack || err));
|
||||
process.exitCode = 1;
|
||||
});
|
||||
@@ -1,50 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-slippage-worker-drift
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "6"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-slippage-worker-drift
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-slippage-worker-drift
|
||||
spec:
|
||||
containers:
|
||||
- name: worker
|
||||
image: node:20-slim
|
||||
imagePullPolicy: IfNotPresent
|
||||
env:
|
||||
- name: HASURA_GRAPHQL_URL
|
||||
value: http://hasura:8080/v1/graphql
|
||||
- name: HASURA_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-hasura
|
||||
key: HASURA_GRAPHQL_ADMIN_SECRET
|
||||
- name: DLOB_SOURCE
|
||||
value: drift
|
||||
- name: DLOB_MARKETS
|
||||
value: SOL-PERP,DOGE-PERP,JUP-PERP
|
||||
- name: DLOB_POLL_MS
|
||||
value: "1000"
|
||||
- name: DLOB_SLIPPAGE_SIZES_USD
|
||||
value: "0.1,0.2,0.5,1,2,5,10,25,50,100,250,500,1000,5000,10000,50000"
|
||||
- name: PRICE_PRECISION
|
||||
value: "1000000"
|
||||
- name: BASE_PRECISION
|
||||
value: "1000000000"
|
||||
command: ["node", "/app/worker.mjs"]
|
||||
volumeMounts:
|
||||
- name: script
|
||||
mountPath: /app/worker.mjs
|
||||
subPath: worker.mjs
|
||||
readOnly: true
|
||||
volumes:
|
||||
- name: script
|
||||
configMap:
|
||||
name: dlob-slippage-worker-script
|
||||
@@ -1,50 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-slippage-worker
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "6"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-slippage-worker
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-slippage-worker
|
||||
spec:
|
||||
containers:
|
||||
- name: worker
|
||||
image: node:20-slim
|
||||
imagePullPolicy: IfNotPresent
|
||||
env:
|
||||
- name: HASURA_GRAPHQL_URL
|
||||
value: http://hasura:8080/v1/graphql
|
||||
- name: HASURA_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-hasura
|
||||
key: HASURA_GRAPHQL_ADMIN_SECRET
|
||||
- name: DLOB_SOURCE
|
||||
value: mevnode
|
||||
- name: DLOB_MARKETS
|
||||
value: SOL-PERP,DOGE-PERP,JUP-PERP
|
||||
- name: DLOB_POLL_MS
|
||||
value: "1000"
|
||||
- name: DLOB_SLIPPAGE_SIZES_USD
|
||||
value: "0.1,0.2,0.5,1,2,5,10,25,50,100,250,500,1000,5000,10000,50000"
|
||||
- name: PRICE_PRECISION
|
||||
value: "1000000"
|
||||
- name: BASE_PRECISION
|
||||
value: "1000000000"
|
||||
command: ["node", "/app/worker.mjs"]
|
||||
volumeMounts:
|
||||
- name: script
|
||||
mountPath: /app/worker.mjs
|
||||
subPath: worker.mjs
|
||||
readOnly: true
|
||||
volumes:
|
||||
- name: script
|
||||
configMap:
|
||||
name: dlob-slippage-worker-script
|
||||
@@ -1,404 +0,0 @@
|
||||
import fs from 'node:fs';
|
||||
import process from 'node:process';
|
||||
import { setTimeout as sleep } from 'node:timers/promises';
|
||||
|
||||
function readJsonFile(filePath) {
|
||||
try {
|
||||
const raw = fs.readFileSync(filePath, 'utf8');
|
||||
return JSON.parse(raw);
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function getIsoNow() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function clampInt(value, min, max, fallback) {
|
||||
const n = Number.parseInt(String(value ?? ''), 10);
|
||||
if (!Number.isInteger(n)) return fallback;
|
||||
return Math.min(max, Math.max(min, n));
|
||||
}
|
||||
|
||||
function envList(name, fallbackCsv) {
|
||||
const raw = process.env[name] ?? fallbackCsv;
|
||||
return String(raw)
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
function parsePositiveNumber(value) {
|
||||
const n = Number.parseFloat(String(value ?? '').trim());
|
||||
if (!Number.isFinite(n) || !(n > 0)) return null;
|
||||
return n;
|
||||
}
|
||||
|
||||
function resolveConfig() {
|
||||
const tokensPath =
|
||||
process.env.HASURA_TOKENS_FILE ||
|
||||
process.env.TOKENS_FILE ||
|
||||
process.env.HASURA_CONFIG_FILE ||
|
||||
'/app/tokens/hasura.json';
|
||||
const tokens = readJsonFile(tokensPath) || {};
|
||||
|
||||
const hasuraUrl =
|
||||
process.env.HASURA_GRAPHQL_URL ||
|
||||
tokens.graphqlUrl ||
|
||||
tokens.apiUrl ||
|
||||
'http://hasura:8080/v1/graphql';
|
||||
const hasuraAdminSecret =
|
||||
process.env.HASURA_ADMIN_SECRET ||
|
||||
process.env.HASURA_GRAPHQL_ADMIN_SECRET ||
|
||||
tokens.adminSecret ||
|
||||
tokens.hasuraAdminSecret;
|
||||
const hasuraAuthToken = process.env.HASURA_AUTH_TOKEN || process.env.HASURA_JWT || undefined;
|
||||
|
||||
const dlobSource = String(process.env.DLOB_SOURCE || 'mevnode').trim() || 'mevnode';
|
||||
const markets = envList('DLOB_MARKETS', 'SOL-PERP,DOGE-PERP,JUP-PERP');
|
||||
const pollMs = clampInt(process.env.DLOB_POLL_MS, 250, 60_000, 1000);
|
||||
|
||||
const sizesUsd = envList('DLOB_SLIPPAGE_SIZES_USD', '10,25,50,100,250,500,1000')
|
||||
.map(parsePositiveNumber)
|
||||
.filter((n) => n != null)
|
||||
.map((n) => n)
|
||||
.filter((n, idx, arr) => arr.findIndex((x) => x === n) === idx)
|
||||
.sort((a, b) => a - b);
|
||||
|
||||
const sizesUsdInt = sizesUsd.filter((n) => Number.isInteger(n));
|
||||
|
||||
const depthLevels = clampInt(process.env.DLOB_DEPTH, 1, 50, 25);
|
||||
const pricePrecision = Number(process.env.PRICE_PRECISION || 1_000_000);
|
||||
const basePrecision = Number(process.env.BASE_PRECISION || 1_000_000_000);
|
||||
if (!Number.isFinite(pricePrecision) || pricePrecision <= 0) throw new Error(`Invalid PRICE_PRECISION: ${process.env.PRICE_PRECISION}`);
|
||||
if (!Number.isFinite(basePrecision) || basePrecision <= 0) throw new Error(`Invalid BASE_PRECISION: ${process.env.BASE_PRECISION}`);
|
||||
|
||||
return {
|
||||
hasuraUrl,
|
||||
hasuraAdminSecret,
|
||||
hasuraAuthToken,
|
||||
dlobSource,
|
||||
markets,
|
||||
pollMs,
|
||||
sizesUsd,
|
||||
sizesUsdInt,
|
||||
depthLevels,
|
||||
pricePrecision,
|
||||
basePrecision,
|
||||
};
|
||||
}
|
||||
|
||||
async function graphqlRequest(cfg, query, variables) {
|
||||
const headers = { 'content-type': 'application/json' };
|
||||
if (cfg.hasuraAuthToken) {
|
||||
headers.authorization = `Bearer ${cfg.hasuraAuthToken}`;
|
||||
} else if (cfg.hasuraAdminSecret) {
|
||||
headers['x-hasura-admin-secret'] = cfg.hasuraAdminSecret;
|
||||
} else {
|
||||
throw new Error('Missing Hasura auth (set HASURA_AUTH_TOKEN or HASURA_ADMIN_SECRET or mount tokens/hasura.json)');
|
||||
}
|
||||
|
||||
const res = await fetch(cfg.hasuraUrl, {
|
||||
method: 'POST',
|
||||
headers,
|
||||
body: JSON.stringify({ query, variables }),
|
||||
signal: AbortSignal.timeout(15_000),
|
||||
});
|
||||
const text = await res.text();
|
||||
if (!res.ok) throw new Error(`Hasura HTTP ${res.status}: ${text}`);
|
||||
const json = JSON.parse(text);
|
||||
if (json.errors?.length) {
|
||||
throw new Error(json.errors.map((e) => e.message).join(' | '));
|
||||
}
|
||||
return json.data;
|
||||
}
|
||||
|
||||
function toNumberOrNull(value) {
|
||||
if (value == null) return null;
|
||||
if (typeof value === 'number') return Number.isFinite(value) ? value : null;
|
||||
if (typeof value === 'string') {
|
||||
const s = value.trim();
|
||||
if (!s) return null;
|
||||
const n = Number(s);
|
||||
return Number.isFinite(n) ? n : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function normalizeLevels(raw) {
|
||||
if (raw == null) return [];
|
||||
if (Array.isArray(raw)) return raw;
|
||||
if (typeof raw === 'string') {
|
||||
const s = raw.trim();
|
||||
if (!s) return [];
|
||||
try {
|
||||
const v = JSON.parse(s);
|
||||
return Array.isArray(v) ? v : [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
function parseScaledLevels(raw, pricePrecision, basePrecision) {
|
||||
const levels = normalizeLevels(raw);
|
||||
const out = [];
|
||||
for (const it of levels) {
|
||||
const priceInt = toNumberOrNull(it?.price);
|
||||
const sizeInt = toNumberOrNull(it?.size);
|
||||
if (priceInt == null || sizeInt == null) continue;
|
||||
const price = priceInt / pricePrecision;
|
||||
const base = sizeInt / basePrecision;
|
||||
if (!Number.isFinite(price) || !Number.isFinite(base)) continue;
|
||||
out.push({ price, base });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function simulateFill(levels, sizeUsd) {
|
||||
let remainingUsd = sizeUsd;
|
||||
let filledUsd = 0;
|
||||
let filledBase = 0;
|
||||
let totalQuoteUsd = 0;
|
||||
let worstPrice = null;
|
||||
let levelsConsumed = 0;
|
||||
|
||||
for (const l of levels) {
|
||||
if (remainingUsd <= 0) break;
|
||||
const levelUsd = l.base * l.price;
|
||||
if (levelUsd <= 0) continue;
|
||||
levelsConsumed += 1;
|
||||
worstPrice = l.price;
|
||||
|
||||
const takeUsd = Math.min(remainingUsd, levelUsd);
|
||||
const takeBase = takeUsd / l.price;
|
||||
|
||||
remainingUsd -= takeUsd;
|
||||
filledUsd += takeUsd;
|
||||
filledBase += takeBase;
|
||||
totalQuoteUsd += takeUsd;
|
||||
}
|
||||
|
||||
const vwapPrice = filledBase > 0 ? totalQuoteUsd / filledBase : null;
|
||||
const fillPct = sizeUsd > 0 ? (filledUsd / sizeUsd) * 100 : null;
|
||||
|
||||
return {
|
||||
filledUsd,
|
||||
filledBase,
|
||||
vwapPrice,
|
||||
worstPrice,
|
||||
levelsConsumed,
|
||||
fillPct,
|
||||
};
|
||||
}
|
||||
|
||||
function impactBps({ side, mid, vwap }) {
|
||||
if (mid == null || vwap == null || mid <= 0) return null;
|
||||
if (side === 'buy') return ((vwap / mid) - 1) * 10_000;
|
||||
if (side === 'sell') return (1 - (vwap / mid)) * 10_000;
|
||||
return null;
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const cfg = resolveConfig();
|
||||
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
service: 'dlob-slippage-worker',
|
||||
startedAt: getIsoNow(),
|
||||
hasuraUrl: cfg.hasuraUrl,
|
||||
hasuraAuth: cfg.hasuraAuthToken ? 'bearer' : cfg.hasuraAdminSecret ? 'admin-secret' : 'none',
|
||||
dlobSource: cfg.dlobSource,
|
||||
markets: cfg.markets,
|
||||
pollMs: cfg.pollMs,
|
||||
sizesUsd: cfg.sizesUsd,
|
||||
depthLevels: cfg.depthLevels,
|
||||
},
|
||||
null,
|
||||
2
|
||||
)
|
||||
);
|
||||
|
||||
const lastSeenUpdatedAt = new Map(); // market -> updated_at
|
||||
|
||||
while (true) {
|
||||
const updatedAt = getIsoNow();
|
||||
|
||||
try {
|
||||
const query = `
|
||||
query DlobL2Latest($source: String!, $markets: [String!]!) {
|
||||
dlob_l2_latest(where: { source: { _eq: $source }, market_name: { _in: $markets } }) {
|
||||
source
|
||||
market_name
|
||||
market_type
|
||||
market_index
|
||||
ts
|
||||
slot
|
||||
best_bid_price
|
||||
best_ask_price
|
||||
bids
|
||||
asks
|
||||
updated_at
|
||||
}
|
||||
}
|
||||
`;
|
||||
|
||||
const data = await graphqlRequest(cfg, query, { source: cfg.dlobSource, markets: cfg.markets });
|
||||
const rows = Array.isArray(data?.dlob_l2_latest) ? data.dlob_l2_latest : [];
|
||||
|
||||
const objectsV1 = [];
|
||||
const objectsV2 = [];
|
||||
|
||||
for (const row of rows) {
|
||||
const market = String(row?.market_name || '').trim();
|
||||
if (!market) continue;
|
||||
|
||||
const rowUpdatedAt = row?.updated_at ?? null;
|
||||
if (rowUpdatedAt && lastSeenUpdatedAt.get(market) === rowUpdatedAt) continue;
|
||||
if (rowUpdatedAt) lastSeenUpdatedAt.set(market, rowUpdatedAt);
|
||||
|
||||
const bestBid = toNumberOrNull(row?.best_bid_price);
|
||||
const bestAsk = toNumberOrNull(row?.best_ask_price);
|
||||
if (bestBid == null || bestAsk == null) continue;
|
||||
|
||||
const mid = (bestBid + bestAsk) / 2;
|
||||
if (!Number.isFinite(mid) || mid <= 0) continue;
|
||||
|
||||
const bids = parseScaledLevels(row?.bids, cfg.pricePrecision, cfg.basePrecision)
|
||||
.slice()
|
||||
.sort((a, b) => b.price - a.price)
|
||||
.slice(0, cfg.depthLevels);
|
||||
const asks = parseScaledLevels(row?.asks, cfg.pricePrecision, cfg.basePrecision)
|
||||
.slice()
|
||||
.sort((a, b) => a.price - b.price)
|
||||
.slice(0, cfg.depthLevels);
|
||||
|
||||
for (const sizeUsd of cfg.sizesUsd) {
|
||||
// buy consumes asks (worse prices as you go up)
|
||||
{
|
||||
const sim = simulateFill(asks, sizeUsd);
|
||||
const baseObj = {
|
||||
source: cfg.dlobSource,
|
||||
market_name: market,
|
||||
side: 'buy',
|
||||
market_type: row?.market_type ?? 'perp',
|
||||
market_index: row?.market_index ?? null,
|
||||
ts: row?.ts == null ? null : String(row.ts),
|
||||
slot: row?.slot == null ? null : String(row.slot),
|
||||
mid_price: String(mid),
|
||||
vwap_price: sim.vwapPrice == null ? null : String(sim.vwapPrice),
|
||||
worst_price: sim.worstPrice == null ? null : String(sim.worstPrice),
|
||||
filled_usd: String(sim.filledUsd),
|
||||
filled_base: String(sim.filledBase),
|
||||
impact_bps: impactBps({ side: 'buy', mid, vwap: sim.vwapPrice }),
|
||||
levels_consumed: sim.levelsConsumed,
|
||||
fill_pct: sim.fillPct == null ? null : String(sim.fillPct),
|
||||
raw: { depthLevels: cfg.depthLevels },
|
||||
updated_at: updatedAt,
|
||||
};
|
||||
objectsV2.push({ ...baseObj, size_usd: String(sizeUsd) });
|
||||
if (Number.isInteger(sizeUsd)) objectsV1.push({ ...baseObj, size_usd: Math.trunc(sizeUsd) });
|
||||
}
|
||||
|
||||
// sell consumes bids (worse prices as you go down)
|
||||
{
|
||||
const sim = simulateFill(bids, sizeUsd);
|
||||
const baseObj = {
|
||||
source: cfg.dlobSource,
|
||||
market_name: market,
|
||||
side: 'sell',
|
||||
market_type: row?.market_type ?? 'perp',
|
||||
market_index: row?.market_index ?? null,
|
||||
ts: row?.ts == null ? null : String(row.ts),
|
||||
slot: row?.slot == null ? null : String(row.slot),
|
||||
mid_price: String(mid),
|
||||
vwap_price: sim.vwapPrice == null ? null : String(sim.vwapPrice),
|
||||
worst_price: sim.worstPrice == null ? null : String(sim.worstPrice),
|
||||
filled_usd: String(sim.filledUsd),
|
||||
filled_base: String(sim.filledBase),
|
||||
impact_bps: impactBps({ side: 'sell', mid, vwap: sim.vwapPrice }),
|
||||
levels_consumed: sim.levelsConsumed,
|
||||
fill_pct: sim.fillPct == null ? null : String(sim.fillPct),
|
||||
raw: { depthLevels: cfg.depthLevels },
|
||||
updated_at: updatedAt,
|
||||
};
|
||||
objectsV2.push({ ...baseObj, size_usd: String(sizeUsd) });
|
||||
if (Number.isInteger(sizeUsd)) objectsV1.push({ ...baseObj, size_usd: Math.trunc(sizeUsd) });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (objectsV1.length) {
|
||||
const mutation = `
|
||||
mutation UpsertSlippageV1($rows: [dlob_slippage_latest_insert_input!]!) {
|
||||
insert_dlob_slippage_latest(
|
||||
objects: $rows
|
||||
on_conflict: {
|
||||
constraint: dlob_slippage_latest_pkey
|
||||
update_columns: [
|
||||
market_type
|
||||
market_index
|
||||
ts
|
||||
slot
|
||||
mid_price
|
||||
vwap_price
|
||||
worst_price
|
||||
filled_usd
|
||||
filled_base
|
||||
impact_bps
|
||||
levels_consumed
|
||||
fill_pct
|
||||
raw
|
||||
updated_at
|
||||
]
|
||||
}
|
||||
) { affected_rows }
|
||||
}
|
||||
`;
|
||||
await graphqlRequest(cfg, mutation, { rows: objectsV1 });
|
||||
}
|
||||
|
||||
if (objectsV2.length) {
|
||||
const mutation = `
|
||||
mutation UpsertSlippageV2($rows: [dlob_slippage_latest_v2_insert_input!]!) {
|
||||
insert_dlob_slippage_latest_v2(
|
||||
objects: $rows
|
||||
on_conflict: {
|
||||
constraint: dlob_slippage_latest_v2_pkey
|
||||
update_columns: [
|
||||
market_type
|
||||
market_index
|
||||
ts
|
||||
slot
|
||||
mid_price
|
||||
vwap_price
|
||||
worst_price
|
||||
filled_usd
|
||||
filled_base
|
||||
impact_bps
|
||||
levels_consumed
|
||||
fill_pct
|
||||
raw
|
||||
updated_at
|
||||
]
|
||||
}
|
||||
) { affected_rows }
|
||||
}
|
||||
`;
|
||||
await graphqlRequest(cfg, mutation, { rows: objectsV2 });
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`[dlob-slippage-worker] ${String(err?.message || err)}`);
|
||||
}
|
||||
|
||||
await sleep(cfg.pollMs);
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(String(err?.stack || err));
|
||||
process.exitCode = 1;
|
||||
});
|
||||
@@ -1,52 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-worker-drift
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "5"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-worker-drift
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-worker-drift
|
||||
spec:
|
||||
hostNetwork: true
|
||||
dnsPolicy: ClusterFirstWithHostNet
|
||||
containers:
|
||||
- name: worker
|
||||
image: node:20-slim
|
||||
imagePullPolicy: IfNotPresent
|
||||
env:
|
||||
- name: HASURA_GRAPHQL_URL
|
||||
value: http://hasura:8080/v1/graphql
|
||||
- name: HASURA_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-hasura
|
||||
key: HASURA_GRAPHQL_ADMIN_SECRET
|
||||
- name: DLOB_SOURCE
|
||||
value: drift
|
||||
- name: DLOB_HTTP_URL
|
||||
value: https://dlob.drift.trade
|
||||
- name: DLOB_FORCE_IPV6
|
||||
value: "true"
|
||||
- name: DLOB_MARKETS
|
||||
value: SOL-PERP,DOGE-PERP,JUP-PERP
|
||||
- name: DLOB_POLL_MS
|
||||
value: "500"
|
||||
- name: DLOB_DEPTH
|
||||
value: "10"
|
||||
command: ["node", "/app/worker.mjs"]
|
||||
volumeMounts:
|
||||
- name: script
|
||||
mountPath: /app/worker.mjs
|
||||
subPath: worker.mjs
|
||||
readOnly: true
|
||||
volumes:
|
||||
- name: script
|
||||
configMap:
|
||||
name: dlob-worker-script
|
||||
@@ -1,48 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-worker
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "5"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-worker
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-worker
|
||||
spec:
|
||||
containers:
|
||||
- name: worker
|
||||
image: node:20-slim
|
||||
imagePullPolicy: IfNotPresent
|
||||
env:
|
||||
- name: HASURA_GRAPHQL_URL
|
||||
value: http://hasura:8080/v1/graphql
|
||||
- name: HASURA_ADMIN_SECRET
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-hasura
|
||||
key: HASURA_GRAPHQL_ADMIN_SECRET
|
||||
- name: DLOB_SOURCE
|
||||
value: mevnode
|
||||
- name: DLOB_HTTP_URL
|
||||
value: http://dlob-server:6969
|
||||
- name: DLOB_MARKETS
|
||||
value: SOL-PERP,DOGE-PERP,JUP-PERP
|
||||
- name: DLOB_POLL_MS
|
||||
value: "500"
|
||||
- name: DLOB_DEPTH
|
||||
value: "10"
|
||||
command: ["node", "/app/worker.mjs"]
|
||||
volumeMounts:
|
||||
- name: script
|
||||
mountPath: /app/worker.mjs
|
||||
subPath: worker.mjs
|
||||
readOnly: true
|
||||
volumes:
|
||||
- name: script
|
||||
configMap:
|
||||
name: dlob-worker-script
|
||||
@@ -1,435 +0,0 @@
|
||||
import fs from 'node:fs';
|
||||
import * as http from 'node:http';
|
||||
import * as https from 'node:https';
|
||||
import process from 'node:process';
|
||||
import { setTimeout as sleep } from 'node:timers/promises';
|
||||
|
||||
function readJsonFile(filePath) {
|
||||
try {
|
||||
const raw = fs.readFileSync(filePath, 'utf8');
|
||||
return JSON.parse(raw);
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function getIsoNow() {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function clampInt(value, min, max, fallback) {
|
||||
const n = Number.parseInt(String(value ?? ''), 10);
|
||||
if (!Number.isInteger(n)) return fallback;
|
||||
return Math.min(max, Math.max(min, n));
|
||||
}
|
||||
|
||||
function envList(name, fallbackCsv) {
|
||||
const raw = process.env[name] ?? fallbackCsv;
|
||||
return String(raw)
|
||||
.split(',')
|
||||
.map((s) => s.trim())
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
function envBool(name, fallback = false) {
|
||||
const raw = process.env[name];
|
||||
if (raw == null) return fallback;
|
||||
const v = String(raw).trim().toLowerCase();
|
||||
if (['1', 'true', 'yes', 'y', 'on'].includes(v)) return true;
|
||||
if (['0', 'false', 'no', 'n', 'off'].includes(v)) return false;
|
||||
return fallback;
|
||||
}
|
||||
|
||||
function resolveConfig() {
|
||||
const tokensPath =
|
||||
process.env.HASURA_TOKENS_FILE ||
|
||||
process.env.TOKENS_FILE ||
|
||||
process.env.HASURA_CONFIG_FILE ||
|
||||
'/app/tokens/hasura.json';
|
||||
const tokens = readJsonFile(tokensPath) || {};
|
||||
|
||||
const hasuraUrl =
|
||||
process.env.HASURA_GRAPHQL_URL ||
|
||||
tokens.graphqlUrl ||
|
||||
tokens.apiUrl ||
|
||||
'http://hasura:8080/v1/graphql';
|
||||
const hasuraAdminSecret =
|
||||
process.env.HASURA_ADMIN_SECRET ||
|
||||
process.env.HASURA_GRAPHQL_ADMIN_SECRET ||
|
||||
tokens.adminSecret ||
|
||||
tokens.hasuraAdminSecret;
|
||||
const hasuraAuthToken = process.env.HASURA_AUTH_TOKEN || process.env.HASURA_JWT || undefined;
|
||||
|
||||
const dlobHttpBase = String(process.env.DLOB_HTTP_URL || process.env.DLOB_HTTP_BASE || 'https://dlob.drift.trade')
|
||||
.trim()
|
||||
.replace(/\/$/, '');
|
||||
const dlobForceIpv6 = envBool('DLOB_FORCE_IPV6', false);
|
||||
const dlobSource = String(process.env.DLOB_SOURCE || 'mevnode').trim() || 'mevnode';
|
||||
|
||||
const markets = envList('DLOB_MARKETS', 'SOL-PERP,DOGE-PERP,JUP-PERP');
|
||||
const depth = clampInt(process.env.DLOB_DEPTH, 1, 50, 10);
|
||||
const pollMs = clampInt(process.env.DLOB_POLL_MS, 100, 10_000, 500);
|
||||
|
||||
const pricePrecision = Number(process.env.PRICE_PRECISION || 1_000_000);
|
||||
const basePrecision = Number(process.env.BASE_PRECISION || 1_000_000_000);
|
||||
if (!Number.isFinite(pricePrecision) || pricePrecision <= 0)
|
||||
throw new Error(`Invalid PRICE_PRECISION: ${process.env.PRICE_PRECISION}`);
|
||||
if (!Number.isFinite(basePrecision) || basePrecision <= 0)
|
||||
throw new Error(`Invalid BASE_PRECISION: ${process.env.BASE_PRECISION}`);
|
||||
|
||||
return {
|
||||
hasuraUrl,
|
||||
hasuraAdminSecret,
|
||||
hasuraAuthToken,
|
||||
dlobSource,
|
||||
dlobHttpBase,
|
||||
dlobForceIpv6,
|
||||
markets,
|
||||
depth,
|
||||
pollMs,
|
||||
pricePrecision,
|
||||
basePrecision,
|
||||
};
|
||||
}
|
||||
|
||||
async function requestText(url, { timeoutMs, family } = {}) {
|
||||
const u = new URL(url);
|
||||
const client = u.protocol === 'https:' ? https : http;
|
||||
|
||||
const port = u.port ? Number.parseInt(u.port, 10) : u.protocol === 'https:' ? 443 : 80;
|
||||
if (!Number.isFinite(port)) throw new Error(`Invalid port for url: ${url}`);
|
||||
|
||||
return await new Promise((resolve, reject) => {
|
||||
const req = client.request(
|
||||
{
|
||||
protocol: u.protocol,
|
||||
hostname: u.hostname,
|
||||
port,
|
||||
path: `${u.pathname}${u.search}`,
|
||||
method: 'GET',
|
||||
family,
|
||||
servername: u.hostname,
|
||||
headers: {
|
||||
accept: 'application/json',
|
||||
},
|
||||
},
|
||||
(res) => {
|
||||
let data = '';
|
||||
res.setEncoding('utf8');
|
||||
res.on('data', (chunk) => {
|
||||
data += chunk;
|
||||
});
|
||||
res.on('end', () => {
|
||||
resolve({ status: res.statusCode ?? 0, text: data });
|
||||
});
|
||||
}
|
||||
);
|
||||
|
||||
req.on('error', reject);
|
||||
req.setTimeout(timeoutMs ?? 5_000, () => {
|
||||
req.destroy(new Error(`Timeout after ${timeoutMs ?? 5_000}ms`));
|
||||
});
|
||||
req.end();
|
||||
});
|
||||
}
|
||||
|
||||
async function graphqlRequest(cfg, query, variables) {
|
||||
const headers = { 'content-type': 'application/json' };
|
||||
if (cfg.hasuraAuthToken) {
|
||||
headers.authorization = `Bearer ${cfg.hasuraAuthToken}`;
|
||||
} else if (cfg.hasuraAdminSecret) {
|
||||
headers['x-hasura-admin-secret'] = cfg.hasuraAdminSecret;
|
||||
} else {
|
||||
throw new Error('Missing Hasura auth (set HASURA_AUTH_TOKEN or HASURA_ADMIN_SECRET or mount tokens/hasura.json)');
|
||||
}
|
||||
|
||||
const res = await fetch(cfg.hasuraUrl, {
|
||||
method: 'POST',
|
||||
headers,
|
||||
body: JSON.stringify({ query, variables }),
|
||||
signal: AbortSignal.timeout(10_000),
|
||||
});
|
||||
const text = await res.text();
|
||||
if (!res.ok) throw new Error(`Hasura HTTP ${res.status}: ${text}`);
|
||||
const json = JSON.parse(text);
|
||||
if (json.errors?.length) {
|
||||
throw new Error(json.errors.map((e) => e.message).join(' | '));
|
||||
}
|
||||
return json.data;
|
||||
}
|
||||
|
||||
function toNumberOrNull(value) {
|
||||
if (value == null) return null;
|
||||
if (typeof value === 'number') return Number.isFinite(value) ? value : null;
|
||||
if (typeof value === 'string') {
|
||||
const s = value.trim();
|
||||
if (!s) return null;
|
||||
const n = Number(s);
|
||||
return Number.isFinite(n) ? n : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function numStr(value) {
|
||||
if (value == null) return null;
|
||||
if (typeof value === 'number') return Number.isFinite(value) ? String(value) : null;
|
||||
if (typeof value === 'string') return value.trim() || null;
|
||||
return null;
|
||||
}
|
||||
|
||||
function parseScaled(valueRaw, scale) {
|
||||
const n = toNumberOrNull(valueRaw);
|
||||
if (n == null) return null;
|
||||
return n / scale;
|
||||
}
|
||||
|
||||
function computeStats({ l2, depth, pricePrecision, basePrecision }) {
|
||||
const bids = Array.isArray(l2?.bids) ? l2.bids : [];
|
||||
const asks = Array.isArray(l2?.asks) ? l2.asks : [];
|
||||
|
||||
const bestBid = parseScaled(l2?.bestBidPrice ?? bids?.[0]?.price, pricePrecision);
|
||||
const bestAsk = parseScaled(l2?.bestAskPrice ?? asks?.[0]?.price, pricePrecision);
|
||||
const markPrice = parseScaled(l2?.markPrice, pricePrecision);
|
||||
const oraclePrice = parseScaled(l2?.oracleData?.price ?? l2?.oracle, pricePrecision);
|
||||
|
||||
const mid = bestBid != null && bestAsk != null ? (bestBid + bestAsk) / 2 : null;
|
||||
const spreadAbs = bestBid != null && bestAsk != null ? bestAsk - bestBid : null;
|
||||
const spreadBps = spreadAbs != null && mid != null && mid > 0 ? (spreadAbs / mid) * 10_000 : null;
|
||||
|
||||
const levels = Math.max(1, depth);
|
||||
let bidBase = 0;
|
||||
let askBase = 0;
|
||||
let bidUsd = 0;
|
||||
let askUsd = 0;
|
||||
|
||||
for (let i = 0; i < Math.min(levels, bids.length); i += 1) {
|
||||
const p = parseScaled(bids[i]?.price, pricePrecision);
|
||||
const s = toNumberOrNull(bids[i]?.size);
|
||||
if (p == null || s == null) continue;
|
||||
const base = s / basePrecision;
|
||||
bidBase += base;
|
||||
bidUsd += base * p;
|
||||
}
|
||||
|
||||
for (let i = 0; i < Math.min(levels, asks.length); i += 1) {
|
||||
const p = parseScaled(asks[i]?.price, pricePrecision);
|
||||
const s = toNumberOrNull(asks[i]?.size);
|
||||
if (p == null || s == null) continue;
|
||||
const base = s / basePrecision;
|
||||
askBase += base;
|
||||
askUsd += base * p;
|
||||
}
|
||||
|
||||
const denom = bidUsd + askUsd;
|
||||
const imbalance = denom > 0 ? (bidUsd - askUsd) / denom : null;
|
||||
|
||||
return {
|
||||
bestBid,
|
||||
bestAsk,
|
||||
mid,
|
||||
spreadAbs,
|
||||
spreadBps,
|
||||
markPrice,
|
||||
oraclePrice,
|
||||
depthLevels: levels,
|
||||
bidBase,
|
||||
askBase,
|
||||
bidUsd,
|
||||
askUsd,
|
||||
imbalance,
|
||||
};
|
||||
}
|
||||
|
||||
function l2ToInsertObject({ dlobSource, l2, updatedAt, pricePrecision }) {
|
||||
return {
|
||||
source: dlobSource,
|
||||
market_name: String(l2.marketName),
|
||||
market_type: String(l2.marketType || 'perp'),
|
||||
market_index: typeof l2.marketIndex === 'number' ? l2.marketIndex : null,
|
||||
ts: l2.ts == null ? null : String(l2.ts),
|
||||
slot: l2.slot == null ? null : String(l2.slot),
|
||||
mark_price: numStr(parseScaled(l2.markPrice, pricePrecision)),
|
||||
oracle_price: numStr(parseScaled(l2.oracleData?.price ?? l2.oracle, pricePrecision)),
|
||||
best_bid_price: numStr(parseScaled(l2.bestBidPrice, pricePrecision)),
|
||||
best_ask_price: numStr(parseScaled(l2.bestAskPrice, pricePrecision)),
|
||||
bids: l2.bids ?? null,
|
||||
asks: l2.asks ?? null,
|
||||
raw: l2 ?? null,
|
||||
updated_at: updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function statsToInsertObject({ dlobSource, l2, stats, updatedAt }) {
|
||||
return {
|
||||
source: dlobSource,
|
||||
market_name: String(l2.marketName),
|
||||
market_type: String(l2.marketType || 'perp'),
|
||||
market_index: typeof l2.marketIndex === 'number' ? l2.marketIndex : null,
|
||||
ts: l2.ts == null ? null : String(l2.ts),
|
||||
slot: l2.slot == null ? null : String(l2.slot),
|
||||
mark_price: stats.markPrice == null ? null : String(stats.markPrice),
|
||||
oracle_price: stats.oraclePrice == null ? null : String(stats.oraclePrice),
|
||||
best_bid_price: stats.bestBid == null ? null : String(stats.bestBid),
|
||||
best_ask_price: stats.bestAsk == null ? null : String(stats.bestAsk),
|
||||
mid_price: stats.mid == null ? null : String(stats.mid),
|
||||
spread_abs: stats.spreadAbs == null ? null : String(stats.spreadAbs),
|
||||
spread_bps: stats.spreadBps == null ? null : String(stats.spreadBps),
|
||||
depth_levels: stats.depthLevels,
|
||||
depth_bid_base: Number.isFinite(stats.bidBase) ? String(stats.bidBase) : null,
|
||||
depth_ask_base: Number.isFinite(stats.askBase) ? String(stats.askBase) : null,
|
||||
depth_bid_usd: Number.isFinite(stats.bidUsd) ? String(stats.bidUsd) : null,
|
||||
depth_ask_usd: Number.isFinite(stats.askUsd) ? String(stats.askUsd) : null,
|
||||
imbalance: stats.imbalance == null ? null : String(stats.imbalance),
|
||||
raw: {
|
||||
spreadPct: l2.spreadPct ?? null,
|
||||
spreadQuote: l2.spreadQuote ?? null,
|
||||
},
|
||||
updated_at: updatedAt,
|
||||
};
|
||||
}
|
||||
|
||||
async function fetchL2(cfg, marketName) {
|
||||
const u = new URL(`${cfg.dlobHttpBase}/l2`);
|
||||
u.searchParams.set('marketName', marketName);
|
||||
u.searchParams.set('depth', String(cfg.depth));
|
||||
|
||||
const url = u.toString();
|
||||
if (cfg.dlobForceIpv6) {
|
||||
const { status, text } = await requestText(url, { timeoutMs: 5_000, family: 6 });
|
||||
if (status < 200 || status >= 300) throw new Error(`DLOB HTTP ${status}: ${text}`);
|
||||
return JSON.parse(text);
|
||||
}
|
||||
|
||||
const res = await fetch(url, { signal: AbortSignal.timeout(5_000) });
|
||||
const text = await res.text();
|
||||
if (!res.ok) throw new Error(`DLOB HTTP ${res.status}: ${text}`);
|
||||
return JSON.parse(text);
|
||||
}
|
||||
|
||||
async function upsertBatch(cfg, l2Objects, statsObjects) {
|
||||
if (!l2Objects.length && !statsObjects.length) return;
|
||||
|
||||
const mutation = `
|
||||
mutation UpsertDlob($l2: [dlob_l2_latest_insert_input!]!, $stats: [dlob_stats_latest_insert_input!]!) {
|
||||
insert_dlob_l2_latest(
|
||||
objects: $l2
|
||||
on_conflict: {
|
||||
constraint: dlob_l2_latest_pkey
|
||||
update_columns: [
|
||||
market_type
|
||||
market_index
|
||||
ts
|
||||
slot
|
||||
mark_price
|
||||
oracle_price
|
||||
best_bid_price
|
||||
best_ask_price
|
||||
bids
|
||||
asks
|
||||
raw
|
||||
updated_at
|
||||
]
|
||||
}
|
||||
) { affected_rows }
|
||||
insert_dlob_stats_latest(
|
||||
objects: $stats
|
||||
on_conflict: {
|
||||
constraint: dlob_stats_latest_pkey
|
||||
update_columns: [
|
||||
market_type
|
||||
market_index
|
||||
ts
|
||||
slot
|
||||
mark_price
|
||||
oracle_price
|
||||
best_bid_price
|
||||
best_ask_price
|
||||
mid_price
|
||||
spread_abs
|
||||
spread_bps
|
||||
depth_levels
|
||||
depth_bid_base
|
||||
depth_ask_base
|
||||
depth_bid_usd
|
||||
depth_ask_usd
|
||||
imbalance
|
||||
raw
|
||||
updated_at
|
||||
]
|
||||
}
|
||||
) { affected_rows }
|
||||
}
|
||||
`;
|
||||
|
||||
await graphqlRequest(cfg, mutation, { l2: l2Objects, stats: statsObjects });
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const cfg = resolveConfig();
|
||||
const lastTsByMarket = new Map();
|
||||
|
||||
console.log(
|
||||
JSON.stringify(
|
||||
{
|
||||
service: 'dlob-worker',
|
||||
startedAt: getIsoNow(),
|
||||
hasuraUrl: cfg.hasuraUrl,
|
||||
hasuraAuth: cfg.hasuraAuthToken ? 'bearer' : cfg.hasuraAdminSecret ? 'admin-secret' : 'none',
|
||||
dlobSource: cfg.dlobSource,
|
||||
dlobHttpBase: cfg.dlobHttpBase,
|
||||
dlobForceIpv6: cfg.dlobForceIpv6,
|
||||
markets: cfg.markets,
|
||||
depth: cfg.depth,
|
||||
pollMs: cfg.pollMs,
|
||||
},
|
||||
null,
|
||||
2
|
||||
)
|
||||
);
|
||||
|
||||
while (true) {
|
||||
const updatedAt = getIsoNow();
|
||||
|
||||
const results = await Promise.allSettled(cfg.markets.map((m) => fetchL2(cfg, m)));
|
||||
const l2Objects = [];
|
||||
const statsObjects = [];
|
||||
|
||||
for (let i = 0; i < results.length; i += 1) {
|
||||
const market = cfg.markets[i];
|
||||
const r = results[i];
|
||||
if (r.status !== 'fulfilled') {
|
||||
console.error(`[dlob-worker] fetch ${market}: ${String(r.reason?.message || r.reason)}`);
|
||||
continue;
|
||||
}
|
||||
const l2 = r.value;
|
||||
if (!l2?.marketName) continue;
|
||||
|
||||
const ts = l2.ts == null ? null : String(l2.ts);
|
||||
if (ts != null && lastTsByMarket.get(l2.marketName) === ts) continue;
|
||||
if (ts != null) lastTsByMarket.set(l2.marketName, ts);
|
||||
|
||||
const stats = computeStats({
|
||||
l2,
|
||||
depth: cfg.depth,
|
||||
pricePrecision: cfg.pricePrecision,
|
||||
basePrecision: cfg.basePrecision,
|
||||
});
|
||||
|
||||
l2Objects.push(l2ToInsertObject({ dlobSource: cfg.dlobSource, l2, updatedAt, pricePrecision: cfg.pricePrecision }));
|
||||
statsObjects.push(statsToInsertObject({ dlobSource: cfg.dlobSource, l2, stats, updatedAt }));
|
||||
}
|
||||
|
||||
try {
|
||||
await upsertBatch(cfg, l2Objects, statsObjects);
|
||||
} catch (err) {
|
||||
console.error(`[dlob-worker] upsert: ${String(err?.message || err)}`);
|
||||
}
|
||||
|
||||
await sleep(cfg.pollMs);
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(String(err?.stack || err));
|
||||
process.exitCode = 1;
|
||||
});
|
||||
@@ -1,24 +1,24 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-publisher
|
||||
name: dlob-publisher-hot
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "4"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-publisher
|
||||
app.kubernetes.io/name: dlob-publisher-hot
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-publisher
|
||||
app.kubernetes.io/name: dlob-publisher-hot
|
||||
spec:
|
||||
imagePullSecrets:
|
||||
- name: gitea-registry
|
||||
containers:
|
||||
- name: publisher
|
||||
image: gitea.mpabi.pl/trade/trade-dlob-server:sha-5d55631-canonical
|
||||
image: gitea.mpabi.pl/trade/trade-dlob-server:grpc-reconnect-20260319-023351
|
||||
imagePullPolicy: IfNotPresent
|
||||
ports:
|
||||
- name: http
|
||||
@@ -41,7 +41,7 @@ spec:
|
||||
- name: ELASTICACHE_PORT
|
||||
value: "6379"
|
||||
- name: REDIS_CLIENT
|
||||
value: DLOB
|
||||
value: DLOB_HOT
|
||||
- name: PERP_MARKETS_TO_LOAD
|
||||
value: "0,7,24"
|
||||
- name: ENDPOINT
|
||||
|
||||
@@ -1,63 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: dlob-server
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "4"
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: dlob-server
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: dlob-server
|
||||
spec:
|
||||
imagePullSecrets:
|
||||
- name: gitea-registry
|
||||
containers:
|
||||
- name: server
|
||||
image: gitea.mpabi.pl/trade/trade-dlob-server:sha-5d55631-canonical
|
||||
imagePullPolicy: IfNotPresent
|
||||
ports:
|
||||
- name: http
|
||||
containerPort: 6969
|
||||
env:
|
||||
- name: RUNNING_LOCAL
|
||||
value: "true"
|
||||
- name: LOCAL_CACHE
|
||||
value: "true"
|
||||
- name: ENV
|
||||
value: mainnet-beta
|
||||
- name: PORT
|
||||
value: "6969"
|
||||
- name: ELASTICACHE_HOST
|
||||
value: dlob-redis
|
||||
- name: ELASTICACHE_PORT
|
||||
value: "6379"
|
||||
- name: REDIS_CLIENT
|
||||
value: DLOB
|
||||
- name: ENDPOINT
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-dlob-rpc
|
||||
key: ENDPOINT
|
||||
- name: WS_ENDPOINT
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: trade-dlob-rpc
|
||||
key: WS_ENDPOINT
|
||||
command: ["node", "/lib/serverLite.js"]
|
||||
readinessProbe:
|
||||
httpGet:
|
||||
path: /startup
|
||||
port: http
|
||||
initialDelaySeconds: 5
|
||||
periodSeconds: 10
|
||||
livenessProbe:
|
||||
httpGet:
|
||||
path: /health
|
||||
port: http
|
||||
initialDelaySeconds: 20
|
||||
periodSeconds: 20
|
||||
@@ -1,13 +0,0 @@
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: dlob-server
|
||||
annotations:
|
||||
argocd.argoproj.io/sync-wave: "4"
|
||||
spec:
|
||||
selector:
|
||||
app.kubernetes.io/name: dlob-server
|
||||
ports:
|
||||
- name: http
|
||||
port: 6969
|
||||
targetPort: http
|
||||
@@ -16,7 +16,7 @@ spec:
|
||||
- name: gitea-registry
|
||||
containers:
|
||||
- name: frontend
|
||||
image: gitea.mpabi.pl/trade/trade-frontend:sha-b06fe7f
|
||||
image: gitea.mpabi.pl/trade/trade-frontend:pg-derived-phase1-20260319-030252
|
||||
imagePullPolicy: IfNotPresent
|
||||
ports:
|
||||
- name: http
|
||||
@@ -26,12 +26,8 @@ spec:
|
||||
value: "8081"
|
||||
- name: APP_VERSION
|
||||
value: "staging"
|
||||
- name: API_UPSTREAM
|
||||
value: "http://trade-api:8787"
|
||||
- name: BASIC_AUTH_FILE
|
||||
value: "/tokens/frontend.json"
|
||||
- name: API_READ_TOKEN_FILE
|
||||
value: "/tokens/read.json"
|
||||
volumeMounts:
|
||||
- name: tokens
|
||||
mountPath: /tokens
|
||||
|
||||
@@ -92,25 +92,33 @@ async function main() {
|
||||
const PUBLIC_DLOB_SOURCE_HEADER = 'X-Hasura-Dlob-Source';
|
||||
|
||||
const apiTokensTable = { schema: 'public', name: 'api_tokens' };
|
||||
const botConfigTable = { schema: 'public', name: 'bot_config' };
|
||||
const botStateTable = { schema: 'public', name: 'bot_state' };
|
||||
const botEventsTable = { schema: 'public', name: 'bot_events' };
|
||||
const source = 'default';
|
||||
|
||||
const baseTicks = { schema: 'public', name: 'drift_ticks' };
|
||||
const dlobL2LatestTable = { schema: 'public', name: 'dlob_l2_latest' };
|
||||
const dlobL2LatestProjectionView = { schema: 'public', name: 'dlob_l2_latest_projection' };
|
||||
const dlobL3LatestProjectionView = { schema: 'public', name: 'dlob_l3_latest_projection' };
|
||||
const dlobBestMakersLatestProjectionView = { schema: 'public', name: 'dlob_best_makers_latest_projection' };
|
||||
const dlobStatsLatestTable = { schema: 'public', name: 'dlob_stats_latest' };
|
||||
const dlobStatsLatestProjectionView = { schema: 'public', name: 'dlob_stats_latest_projection' };
|
||||
const dlobDepthBpsLatestTable = { schema: 'public', name: 'dlob_depth_bps_latest' };
|
||||
const dlobSlippageLatestTable = { schema: 'public', name: 'dlob_slippage_latest' };
|
||||
const dlobSlippageLatestV2Table = { schema: 'public', name: 'dlob_slippage_latest_v2' };
|
||||
const dlobHotSnapshotLatestTable = { schema: 'public', name: 'dlob_hot_snapshot_latest' };
|
||||
const dlobHotDerivedLatestTable = { schema: 'public', name: 'dlob_hot_derived_latest' };
|
||||
const dlobAllDerivedLatestTable = { schema: 'public', name: 'dlob_all_derived_latest' };
|
||||
const candlesCacheTable = { schema: 'public', name: 'drift_candles_cache' };
|
||||
const dlobStatsTsTable = { schema: 'public', name: 'dlob_stats_ts' };
|
||||
const dlobDepthBpsTsTable = { schema: 'public', name: 'dlob_depth_bps_ts' };
|
||||
const dlobSlippageTsTable = { schema: 'public', name: 'dlob_slippage_ts' };
|
||||
const dlobSlippageTsV2Table = { schema: 'public', name: 'dlob_slippage_ts_v2' };
|
||||
const dlobHotDerivedTsTable = { schema: 'public', name: 'dlob_hot_derived_ts' };
|
||||
const dlobAllDerivedTsTable = { schema: 'public', name: 'dlob_all_derived_ts' };
|
||||
const baseCandlesFn = { schema: 'public', name: 'get_drift_candles' };
|
||||
const candlesReturnTable = { schema: 'public', name: 'drift_candles' };
|
||||
const legacyDlobRelations = [
|
||||
{ schema: 'public', name: 'dlob_l2_latest' },
|
||||
{ schema: 'public', name: 'dlob_l2_latest_projection' },
|
||||
{ schema: 'public', name: 'dlob_stats_latest' },
|
||||
{ schema: 'public', name: 'dlob_stats_latest_projection' },
|
||||
{ schema: 'public', name: 'dlob_depth_bps_latest' },
|
||||
{ schema: 'public', name: 'dlob_slippage_latest' },
|
||||
{ schema: 'public', name: 'dlob_slippage_latest_v2' },
|
||||
{ schema: 'public', name: 'dlob_stats_ts' },
|
||||
{ schema: 'public', name: 'dlob_depth_bps_ts' },
|
||||
{ schema: 'public', name: 'dlob_slippage_ts' },
|
||||
{ schema: 'public', name: 'dlob_slippage_ts_v2' },
|
||||
];
|
||||
|
||||
const extraTicksName = normalizeName(TARGET_TICKS_TABLE);
|
||||
const extraCandlesName = normalizeName(TARGET_CANDLES_FUNCTION);
|
||||
@@ -293,139 +301,174 @@ async function main() {
|
||||
|
||||
const dlobPublicFilter = { source: { _eq: PUBLIC_DLOB_SOURCE_HEADER } };
|
||||
|
||||
await metadataIgnore({ type: 'pg_untrack_table', args: { source, table: dlobL2LatestTable, cascade: true } });
|
||||
await ensureProjectedPublicSelectTable(dlobL2LatestProjectionView, [
|
||||
for (const table of legacyDlobRelations) {
|
||||
await metadataIgnore({ type: 'pg_untrack_table', args: { source, table, cascade: true } });
|
||||
}
|
||||
|
||||
await ensurePublicSelectTable(dlobHotSnapshotLatestTable, [
|
||||
'source',
|
||||
'market_name',
|
||||
'redis_key',
|
||||
'snapshot_kind',
|
||||
'market_type',
|
||||
'market_index',
|
||||
'ts',
|
||||
'market_name',
|
||||
'is_indicative',
|
||||
'ts_ms',
|
||||
'slot',
|
||||
'mark_price',
|
||||
'oracle_price',
|
||||
'best_bid_price',
|
||||
'best_ask_price',
|
||||
'market_slot',
|
||||
'payload_hash',
|
||||
'mark_price_raw',
|
||||
'oracle_price_raw',
|
||||
'best_bid_price_raw',
|
||||
'best_ask_price_raw',
|
||||
'spread_pct_raw',
|
||||
'spread_quote_raw',
|
||||
'oracle_data',
|
||||
'bids',
|
||||
'asks',
|
||||
'raw',
|
||||
'payload',
|
||||
'updated_at',
|
||||
], { publicFilter: dlobPublicFilter, customName: 'dlob_l2_latest' });
|
||||
]);
|
||||
|
||||
await metadataIgnore({ type: 'pg_untrack_table', args: { source, table: dlobStatsLatestTable, cascade: true } });
|
||||
await ensureProjectedPublicSelectTable(dlobStatsLatestProjectionView, [
|
||||
await ensurePublicSelectTable(dlobHotDerivedLatestTable, [
|
||||
'source',
|
||||
'market_name',
|
||||
'market_type',
|
||||
'market_index',
|
||||
'ts',
|
||||
'market_name',
|
||||
'is_indicative',
|
||||
'ts_ms',
|
||||
'slot',
|
||||
'market_slot',
|
||||
'mark_price',
|
||||
'oracle_price',
|
||||
'best_bid_price',
|
||||
'best_ask_price',
|
||||
'mid_price',
|
||||
'spread_abs',
|
||||
'spread_quote',
|
||||
'spread_bps',
|
||||
'depth_levels',
|
||||
'bid_levels',
|
||||
'ask_levels',
|
||||
'top_bid_size',
|
||||
'top_ask_size',
|
||||
'top_bid_notional',
|
||||
'top_ask_notional',
|
||||
'depth_bid_base',
|
||||
'depth_ask_base',
|
||||
'depth_bid_usd',
|
||||
'depth_ask_usd',
|
||||
'depth_bid_quote',
|
||||
'depth_ask_quote',
|
||||
'imbalance',
|
||||
'raw',
|
||||
'bids_norm',
|
||||
'asks_norm',
|
||||
'raw_payload_hash',
|
||||
'updated_at',
|
||||
], { publicFilter: dlobPublicFilter, customName: 'dlob_stats_latest' });
|
||||
]);
|
||||
|
||||
await ensureProjectedPublicSelectTable(dlobL3LatestProjectionView, [
|
||||
await ensurePublicSelectTable(dlobAllDerivedLatestTable, [
|
||||
'source',
|
||||
'market_name',
|
||||
'market_type',
|
||||
'market_index',
|
||||
'ts',
|
||||
'slot',
|
||||
'bids',
|
||||
'asks',
|
||||
'raw',
|
||||
'updated_at',
|
||||
], { publicFilter: dlobPublicFilter });
|
||||
|
||||
await ensureProjectedPublicSelectTable(dlobBestMakersLatestProjectionView, [
|
||||
'source',
|
||||
'market_name',
|
||||
'market_type',
|
||||
'market_index',
|
||||
'is_indicative',
|
||||
'ts_ms',
|
||||
'slot',
|
||||
'bids',
|
||||
'asks',
|
||||
'raw',
|
||||
'updated_at',
|
||||
], { publicFilter: dlobPublicFilter });
|
||||
|
||||
await ensurePublicSelectTable(dlobDepthBpsLatestTable, [
|
||||
'source',
|
||||
'market_name',
|
||||
'band_bps',
|
||||
'market_type',
|
||||
'market_index',
|
||||
'ts',
|
||||
'slot',
|
||||
'mid_price',
|
||||
'market_slot',
|
||||
'mark_price',
|
||||
'oracle_price',
|
||||
'best_bid_price',
|
||||
'best_ask_price',
|
||||
'bid_base',
|
||||
'ask_base',
|
||||
'bid_usd',
|
||||
'ask_usd',
|
||||
'mid_price',
|
||||
'spread_quote',
|
||||
'spread_bps',
|
||||
'depth_levels',
|
||||
'bid_levels',
|
||||
'ask_levels',
|
||||
'top_bid_size',
|
||||
'top_ask_size',
|
||||
'top_bid_notional',
|
||||
'top_ask_notional',
|
||||
'depth_bid_base',
|
||||
'depth_ask_base',
|
||||
'depth_bid_quote',
|
||||
'depth_ask_quote',
|
||||
'imbalance',
|
||||
'raw',
|
||||
'bids_norm',
|
||||
'asks_norm',
|
||||
'raw_payload_hash',
|
||||
'updated_at',
|
||||
], { publicFilter: dlobPublicFilter });
|
||||
]);
|
||||
|
||||
await ensurePublicSelectTable(dlobSlippageLatestTable, [
|
||||
await ensurePublicSelectTable(dlobHotDerivedTsTable, [
|
||||
'event_ts',
|
||||
'id',
|
||||
'source',
|
||||
'market_name',
|
||||
'side',
|
||||
'size_usd',
|
||||
'market_type',
|
||||
'market_index',
|
||||
'ts',
|
||||
'market_name',
|
||||
'is_indicative',
|
||||
'ts_ms',
|
||||
'slot',
|
||||
'mid_price',
|
||||
'market_slot',
|
||||
'mark_price',
|
||||
'oracle_price',
|
||||
'best_bid_price',
|
||||
'best_ask_price',
|
||||
'vwap_price',
|
||||
'worst_price',
|
||||
'filled_usd',
|
||||
'filled_base',
|
||||
'impact_bps',
|
||||
'levels_consumed',
|
||||
'fill_pct',
|
||||
'raw',
|
||||
'updated_at',
|
||||
], { publicFilter: dlobPublicFilter });
|
||||
'mid_price',
|
||||
'spread_quote',
|
||||
'spread_bps',
|
||||
'depth_levels',
|
||||
'bid_levels',
|
||||
'ask_levels',
|
||||
'top_bid_size',
|
||||
'top_ask_size',
|
||||
'top_bid_notional',
|
||||
'top_ask_notional',
|
||||
'depth_bid_base',
|
||||
'depth_ask_base',
|
||||
'depth_bid_quote',
|
||||
'depth_ask_quote',
|
||||
'imbalance',
|
||||
'bids_norm',
|
||||
'asks_norm',
|
||||
'raw_payload_hash',
|
||||
'inserted_at',
|
||||
]);
|
||||
|
||||
await ensurePublicSelectTable(dlobSlippageLatestV2Table, [
|
||||
await ensurePublicSelectTable(dlobAllDerivedTsTable, [
|
||||
'event_ts',
|
||||
'id',
|
||||
'source',
|
||||
'market_name',
|
||||
'side',
|
||||
'size_usd',
|
||||
'market_type',
|
||||
'market_index',
|
||||
'ts',
|
||||
'market_name',
|
||||
'is_indicative',
|
||||
'ts_ms',
|
||||
'slot',
|
||||
'mid_price',
|
||||
'market_slot',
|
||||
'mark_price',
|
||||
'oracle_price',
|
||||
'best_bid_price',
|
||||
'best_ask_price',
|
||||
'vwap_price',
|
||||
'worst_price',
|
||||
'filled_usd',
|
||||
'filled_base',
|
||||
'impact_bps',
|
||||
'levels_consumed',
|
||||
'fill_pct',
|
||||
'raw',
|
||||
'updated_at',
|
||||
], { publicFilter: dlobPublicFilter });
|
||||
'mid_price',
|
||||
'spread_quote',
|
||||
'spread_bps',
|
||||
'depth_levels',
|
||||
'bid_levels',
|
||||
'ask_levels',
|
||||
'top_bid_size',
|
||||
'top_ask_size',
|
||||
'top_bid_notional',
|
||||
'top_ask_notional',
|
||||
'depth_bid_base',
|
||||
'depth_ask_base',
|
||||
'depth_bid_quote',
|
||||
'depth_ask_quote',
|
||||
'imbalance',
|
||||
'bids_norm',
|
||||
'asks_norm',
|
||||
'raw_payload_hash',
|
||||
'inserted_at',
|
||||
]);
|
||||
|
||||
await ensurePublicSelectTable(dlobStatsTsTable, [
|
||||
'ts',
|
||||
@@ -517,6 +560,12 @@ async function main() {
|
||||
'raw',
|
||||
], { publicFilter: dlobPublicFilter });
|
||||
|
||||
// Bot control-plane tables (tracked; permissions are intentionally not created here by default).
|
||||
for (const t of [botConfigTable, botStateTable, botEventsTable]) {
|
||||
await metadataIgnore({ type: 'pg_untrack_table', args: { source, table: t } });
|
||||
await metadata({ type: 'pg_track_table', args: { source, table: t } });
|
||||
}
|
||||
|
||||
// Return table type for candle functions (needed for Hasura to track the function).
|
||||
await metadataIgnore({ type: 'pg_track_table', args: { source, table: candlesReturnTable } });
|
||||
|
||||
@@ -532,7 +581,7 @@ async function main() {
|
||||
}
|
||||
|
||||
for (const fn of candleFns) {
|
||||
// Function for aggregated candle queries (used by trade-api).
|
||||
// Function for aggregated candle queries and indicator reads over Hasura.
|
||||
await metadataIgnore({ type: 'pg_untrack_function', args: { source, function: fn } });
|
||||
await metadata({ type: 'pg_track_function', args: { source, function: fn } });
|
||||
}
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: trade-ingestor
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/name: trade-ingestor
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: trade-ingestor
|
||||
spec:
|
||||
imagePullSecrets:
|
||||
- name: gitea-registry
|
||||
containers:
|
||||
- name: ingestor
|
||||
image: gitea.mpabi.pl/trade/trade-ingestor:k3s-20260106013603
|
||||
imagePullPolicy: IfNotPresent
|
||||
env:
|
||||
- name: MARKET_NAME
|
||||
value: "PUMP-PERP"
|
||||
- name: INTERVAL_MS
|
||||
value: "1000"
|
||||
- name: SOURCE
|
||||
value: "drift_oracle"
|
||||
- name: INGEST_VIA
|
||||
value: "api"
|
||||
- name: INGEST_API_URL
|
||||
value: "http://trade-api:8787"
|
||||
volumeMounts:
|
||||
- name: tokens
|
||||
mountPath: /app/tokens
|
||||
readOnly: true
|
||||
command:
|
||||
- sh
|
||||
- -c
|
||||
args:
|
||||
- >
|
||||
npm run ingest:oracle -- \
|
||||
--market-name "${MARKET_NAME}" \
|
||||
--interval-ms "${INTERVAL_MS}" \
|
||||
--source "${SOURCE}" \
|
||||
--ingest-via "${INGEST_VIA}"
|
||||
volumes:
|
||||
- name: tokens
|
||||
secret:
|
||||
secretName: trade-ingestor-tokens
|
||||
@@ -111,8 +111,52 @@ ALTER TABLE api_tokens ADD COLUMN IF NOT EXISTS scopes TEXT[] NOT NULL DEFAULT A
|
||||
CREATE INDEX IF NOT EXISTS api_tokens_revoked_at_idx
|
||||
ON api_tokens (revoked_at);
|
||||
|
||||
-- Bot control-plane (desired state) + executor telemetry (state/events).
|
||||
--
|
||||
-- MVP intent:
|
||||
-- - `bot_config`: desired state (mode/market/limits/kill switch)
|
||||
-- - `bot_state`: heartbeat + last error/action
|
||||
-- - `bot_events`: append-only audit log
|
||||
CREATE TABLE IF NOT EXISTS public.bot_config (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
name TEXT NOT NULL UNIQUE,
|
||||
market_name TEXT NOT NULL,
|
||||
market_type TEXT NOT NULL DEFAULT 'perp',
|
||||
mode TEXT NOT NULL DEFAULT 'off',
|
||||
kill_switch BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
params JSONB,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS bot_config_updated_at_desc_idx
|
||||
ON public.bot_config (updated_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.bot_state (
|
||||
bot_id UUID PRIMARY KEY REFERENCES public.bot_config(id) ON DELETE CASCADE,
|
||||
last_heartbeat_at TIMESTAMPTZ,
|
||||
last_action_at TIMESTAMPTZ,
|
||||
last_error TEXT,
|
||||
state JSONB,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS bot_state_updated_at_desc_idx
|
||||
ON public.bot_state (updated_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.bot_events (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
bot_id UUID NOT NULL REFERENCES public.bot_config(id) ON DELETE CASCADE,
|
||||
ts TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
type TEXT NOT NULL,
|
||||
payload JSONB
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS bot_events_bot_ts_desc_idx
|
||||
ON public.bot_events (bot_id, ts DESC);
|
||||
|
||||
-- Compute OHLC candles from `drift_ticks` for a symbol and bucket size.
|
||||
-- Exposed via Hasura (track function) and used by trade-api to compute indicators server-side.
|
||||
-- Exposed via Hasura (track function) for server-side candle aggregation and indicator reads.
|
||||
-- Hasura tracks functions only if they return SETOF a table/view type.
|
||||
-- This table is used purely as the return type for candle functions.
|
||||
CREATE TABLE IF NOT EXISTS public.drift_candles (
|
||||
@@ -838,41 +882,8 @@ CREATE TABLE IF NOT EXISTS public.dlob_orderbook_l2_latest (
|
||||
PRIMARY KEY (source, market_name, is_indicative)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_orderbook_l3_latest (
|
||||
source TEXT NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER,
|
||||
is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
ts BIGINT,
|
||||
slot BIGINT,
|
||||
oracle_data JSONB,
|
||||
bids JSONB,
|
||||
asks JSONB,
|
||||
payload JSONB NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (source, market_name, is_indicative)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_best_makers_latest (
|
||||
source TEXT NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER,
|
||||
slot BIGINT,
|
||||
bids JSONB,
|
||||
asks JSONB,
|
||||
payload JSONB NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (source, market_name)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_orderbook_l2_latest_updated_at_idx
|
||||
ON public.dlob_orderbook_l2_latest (updated_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS dlob_orderbook_l3_latest_updated_at_idx
|
||||
ON public.dlob_orderbook_l3_latest (updated_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS dlob_best_makers_latest_updated_at_idx
|
||||
ON public.dlob_best_makers_latest (updated_at DESC);
|
||||
|
||||
CREATE OR REPLACE VIEW public.dlob_l2_latest_projection AS
|
||||
SELECT
|
||||
@@ -905,34 +916,6 @@ SELECT
|
||||
FROM public.dlob_orderbook_l2_latest
|
||||
WHERE is_indicative = FALSE;
|
||||
|
||||
CREATE OR REPLACE VIEW public.dlob_l3_latest_projection AS
|
||||
SELECT
|
||||
source,
|
||||
market_name,
|
||||
market_type,
|
||||
market_index,
|
||||
ts,
|
||||
slot,
|
||||
bids,
|
||||
asks,
|
||||
payload AS raw,
|
||||
updated_at
|
||||
FROM public.dlob_orderbook_l3_latest
|
||||
WHERE is_indicative = FALSE;
|
||||
|
||||
CREATE OR REPLACE VIEW public.dlob_best_makers_latest_projection AS
|
||||
SELECT
|
||||
source,
|
||||
market_name,
|
||||
market_type,
|
||||
market_index,
|
||||
slot,
|
||||
bids,
|
||||
asks,
|
||||
payload AS raw,
|
||||
updated_at
|
||||
FROM public.dlob_best_makers_latest;
|
||||
|
||||
CREATE OR REPLACE VIEW public.dlob_stats_latest_projection AS
|
||||
WITH base AS (
|
||||
SELECT
|
||||
@@ -1057,3 +1040,286 @@ SELECT
|
||||
FROM base b
|
||||
LEFT JOIN bid_levels bids USING (source, market_name)
|
||||
LEFT JOIN ask_levels asks USING (source, market_name);
|
||||
|
||||
-- Retired legacy DLOB read-side pipeline artifacts. Keep this cleanup in initdb so
|
||||
-- fresh volumes and reused volumes both converge on the current Redis -> derived tables workflow.
|
||||
DROP VIEW IF EXISTS public.dlob_l2_latest_projection;
|
||||
DROP VIEW IF EXISTS public.dlob_stats_latest_projection;
|
||||
DROP TABLE IF EXISTS public.dlob_l2_latest;
|
||||
DROP TABLE IF EXISTS public.dlob_stats_latest;
|
||||
DROP TABLE IF EXISTS public.dlob_depth_bps_latest;
|
||||
DROP TABLE IF EXISTS public.dlob_slippage_latest;
|
||||
DROP TABLE IF EXISTS public.dlob_slippage_latest_v2;
|
||||
DROP TABLE IF EXISTS public.dlob_stats_ts;
|
||||
DROP TABLE IF EXISTS public.dlob_depth_bps_ts;
|
||||
DROP TABLE IF EXISTS public.dlob_slippage_ts;
|
||||
DROP TABLE IF EXISTS public.dlob_slippage_ts_v2;
|
||||
DROP TABLE IF EXISTS public.dlob_orderbook_l2_latest;
|
||||
|
||||
-- Canonical raw DLOB hot snapshots written from Redis to PostgreSQL.
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_hot_snapshot_latest (
|
||||
source TEXT NOT NULL,
|
||||
redis_key TEXT NOT NULL,
|
||||
snapshot_kind TEXT NOT NULL CHECK (snapshot_kind IN ('orderbook_l2', 'orderbook_l3', 'best_makers')),
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
ts_ms BIGINT NOT NULL,
|
||||
slot BIGINT,
|
||||
market_slot BIGINT,
|
||||
payload_hash TEXT NOT NULL,
|
||||
mark_price_raw TEXT,
|
||||
oracle_price_raw TEXT,
|
||||
best_bid_price_raw TEXT,
|
||||
best_ask_price_raw TEXT,
|
||||
spread_pct_raw TEXT,
|
||||
spread_quote_raw TEXT,
|
||||
oracle_data JSONB,
|
||||
bids JSONB,
|
||||
asks JSONB,
|
||||
payload JSONB NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (source, market_type, market_index, snapshot_kind, is_indicative)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_snapshot_latest_updated_at_idx
|
||||
ON public.dlob_hot_snapshot_latest (updated_at DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_snapshot_latest_source_market_idx
|
||||
ON public.dlob_hot_snapshot_latest (source, market_type, market_index, updated_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_hot_snapshot_ts (
|
||||
event_ts TIMESTAMPTZ NOT NULL,
|
||||
id BIGSERIAL NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
redis_key TEXT NOT NULL,
|
||||
snapshot_kind TEXT NOT NULL CHECK (snapshot_kind IN ('orderbook_l2', 'orderbook_l3', 'best_makers')),
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
ts_ms BIGINT NOT NULL,
|
||||
slot BIGINT,
|
||||
market_slot BIGINT,
|
||||
payload_hash TEXT NOT NULL,
|
||||
mark_price_raw TEXT,
|
||||
oracle_price_raw TEXT,
|
||||
best_bid_price_raw TEXT,
|
||||
best_ask_price_raw TEXT,
|
||||
spread_pct_raw TEXT,
|
||||
spread_quote_raw TEXT,
|
||||
oracle_data JSONB,
|
||||
bids JSONB,
|
||||
asks JSONB,
|
||||
payload JSONB NOT NULL,
|
||||
inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (event_ts, id)
|
||||
);
|
||||
|
||||
SELECT create_hypertable('dlob_hot_snapshot_ts', 'event_ts', if_not_exists => TRUE, migrate_data => TRUE);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS dlob_hot_snapshot_ts_dedupe_idx
|
||||
ON public.dlob_hot_snapshot_ts (event_ts, source, redis_key, payload_hash);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_snapshot_ts_market_ts_desc_idx
|
||||
ON public.dlob_hot_snapshot_ts (market_type, market_index, event_ts DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_snapshot_ts_source_market_ts_desc_idx
|
||||
ON public.dlob_hot_snapshot_ts (source, market_type, market_index, event_ts DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_hot_derived_latest (
|
||||
source TEXT NOT NULL,
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
ts_ms BIGINT NOT NULL,
|
||||
slot BIGINT,
|
||||
market_slot BIGINT,
|
||||
mark_price NUMERIC,
|
||||
oracle_price NUMERIC,
|
||||
best_bid_price NUMERIC,
|
||||
best_ask_price NUMERIC,
|
||||
mid_price NUMERIC,
|
||||
spread_quote NUMERIC,
|
||||
spread_bps NUMERIC,
|
||||
depth_levels INTEGER NOT NULL,
|
||||
bid_levels INTEGER NOT NULL,
|
||||
ask_levels INTEGER NOT NULL,
|
||||
top_bid_size NUMERIC,
|
||||
top_ask_size NUMERIC,
|
||||
top_bid_notional NUMERIC,
|
||||
top_ask_notional NUMERIC,
|
||||
depth_bid_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_bid_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
imbalance NUMERIC,
|
||||
bids_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
asks_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
raw_payload_hash TEXT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (source, market_type, market_index, is_indicative)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_derived_latest_updated_at_idx
|
||||
ON public.dlob_hot_derived_latest (updated_at DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_derived_latest_source_market_idx
|
||||
ON public.dlob_hot_derived_latest (source, market_type, market_index, updated_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_hot_derived_ts (
|
||||
event_ts TIMESTAMPTZ NOT NULL,
|
||||
id BIGSERIAL NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
ts_ms BIGINT NOT NULL,
|
||||
slot BIGINT,
|
||||
market_slot BIGINT,
|
||||
mark_price NUMERIC,
|
||||
oracle_price NUMERIC,
|
||||
best_bid_price NUMERIC,
|
||||
best_ask_price NUMERIC,
|
||||
mid_price NUMERIC,
|
||||
spread_quote NUMERIC,
|
||||
spread_bps NUMERIC,
|
||||
depth_levels INTEGER NOT NULL,
|
||||
bid_levels INTEGER NOT NULL,
|
||||
ask_levels INTEGER NOT NULL,
|
||||
top_bid_size NUMERIC,
|
||||
top_ask_size NUMERIC,
|
||||
top_bid_notional NUMERIC,
|
||||
top_ask_notional NUMERIC,
|
||||
depth_bid_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_bid_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
imbalance NUMERIC,
|
||||
bids_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
asks_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
raw_payload_hash TEXT NOT NULL,
|
||||
inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (event_ts, id)
|
||||
);
|
||||
|
||||
SELECT create_hypertable('dlob_hot_derived_ts', 'event_ts', if_not_exists => TRUE, migrate_data => TRUE);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS dlob_hot_derived_ts_dedupe_idx
|
||||
ON public.dlob_hot_derived_ts (event_ts, source, market_type, market_index, is_indicative, raw_payload_hash);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_derived_ts_market_ts_desc_idx
|
||||
ON public.dlob_hot_derived_ts (market_type, market_index, event_ts DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_hot_derived_ts_source_market_ts_desc_idx
|
||||
ON public.dlob_hot_derived_ts (source, market_type, market_index, event_ts DESC);
|
||||
|
||||
DO $$
|
||||
BEGIN
|
||||
PERFORM add_retention_policy('dlob_hot_snapshot_ts', INTERVAL '30 days');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
DO $$
|
||||
BEGIN
|
||||
PERFORM add_retention_policy('dlob_hot_derived_ts', INTERVAL '180 days');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_all_derived_latest (
|
||||
source TEXT NOT NULL,
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
ts_ms BIGINT NOT NULL,
|
||||
slot BIGINT,
|
||||
market_slot BIGINT,
|
||||
mark_price NUMERIC,
|
||||
oracle_price NUMERIC,
|
||||
best_bid_price NUMERIC,
|
||||
best_ask_price NUMERIC,
|
||||
mid_price NUMERIC,
|
||||
spread_quote NUMERIC,
|
||||
spread_bps NUMERIC,
|
||||
depth_levels INTEGER NOT NULL,
|
||||
bid_levels INTEGER NOT NULL,
|
||||
ask_levels INTEGER NOT NULL,
|
||||
top_bid_size NUMERIC,
|
||||
top_ask_size NUMERIC,
|
||||
top_bid_notional NUMERIC,
|
||||
top_ask_notional NUMERIC,
|
||||
depth_bid_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_bid_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
imbalance NUMERIC,
|
||||
bids_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
asks_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
raw_payload_hash TEXT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (source, market_type, market_index, is_indicative)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_all_derived_latest_updated_at_idx
|
||||
ON public.dlob_all_derived_latest (updated_at DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_all_derived_latest_source_market_idx
|
||||
ON public.dlob_all_derived_latest (source, market_type, market_index, updated_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS public.dlob_all_derived_ts (
|
||||
event_ts TIMESTAMPTZ NOT NULL,
|
||||
id BIGSERIAL NOT NULL,
|
||||
source TEXT NOT NULL,
|
||||
market_type TEXT NOT NULL,
|
||||
market_index INTEGER NOT NULL,
|
||||
market_name TEXT NOT NULL,
|
||||
is_indicative BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
ts_ms BIGINT NOT NULL,
|
||||
slot BIGINT,
|
||||
market_slot BIGINT,
|
||||
mark_price NUMERIC,
|
||||
oracle_price NUMERIC,
|
||||
best_bid_price NUMERIC,
|
||||
best_ask_price NUMERIC,
|
||||
mid_price NUMERIC,
|
||||
spread_quote NUMERIC,
|
||||
spread_bps NUMERIC,
|
||||
depth_levels INTEGER NOT NULL,
|
||||
bid_levels INTEGER NOT NULL,
|
||||
ask_levels INTEGER NOT NULL,
|
||||
top_bid_size NUMERIC,
|
||||
top_ask_size NUMERIC,
|
||||
top_bid_notional NUMERIC,
|
||||
top_ask_notional NUMERIC,
|
||||
depth_bid_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_base NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_bid_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
depth_ask_quote NUMERIC NOT NULL DEFAULT 0,
|
||||
imbalance NUMERIC,
|
||||
bids_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
asks_norm JSONB NOT NULL DEFAULT '[]'::jsonb,
|
||||
raw_payload_hash TEXT NOT NULL,
|
||||
inserted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (event_ts, id)
|
||||
);
|
||||
|
||||
SELECT create_hypertable('dlob_all_derived_ts', 'event_ts', if_not_exists => TRUE, migrate_data => TRUE);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS dlob_all_derived_ts_dedupe_idx
|
||||
ON public.dlob_all_derived_ts (event_ts, source, market_type, market_index, is_indicative, raw_payload_hash);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_all_derived_ts_market_ts_desc_idx
|
||||
ON public.dlob_all_derived_ts (market_type, market_index, event_ts DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS dlob_all_derived_ts_source_market_ts_desc_idx
|
||||
ON public.dlob_all_derived_ts (source, market_type, market_index, event_ts DESC);
|
||||
|
||||
DO $$
|
||||
BEGIN
|
||||
PERFORM add_retention_policy('dlob_all_derived_ts', INTERVAL '30 days');
|
||||
EXCEPTION WHEN duplicate_object THEN NULL;
|
||||
END $$;
|
||||
|
||||
@@ -9,23 +9,10 @@ resources:
|
||||
- hasura/service.yaml
|
||||
- hasura/deployment.yaml
|
||||
- hasura/job-bootstrap.yaml
|
||||
- api/service.yaml
|
||||
- api/deployment.yaml
|
||||
- ingestor/deployment.yaml
|
||||
- frontend/service.yaml
|
||||
- frontend/deployment.yaml
|
||||
- dlob/redis.yaml
|
||||
- dlob/publisher-deployment.yaml
|
||||
- dlob/server-service.yaml
|
||||
- dlob/server-deployment.yaml
|
||||
- dlob-worker/deployment.yaml
|
||||
- dlob-worker/deployment-drift.yaml
|
||||
- dlob-depth-worker/deployment.yaml
|
||||
- dlob-depth-worker/deployment-drift.yaml
|
||||
- dlob-slippage-worker/deployment.yaml
|
||||
- dlob-slippage-worker/deployment-drift.yaml
|
||||
- dlob-ts-archiver/deployment.yaml
|
||||
- dlob-ts-archiver/deployment-drift.yaml
|
||||
- candles-cache-worker/deployment.yaml
|
||||
|
||||
configMapGenerator:
|
||||
@@ -35,27 +22,9 @@ configMapGenerator:
|
||||
- name: hasura-bootstrap-script
|
||||
files:
|
||||
- hasura/hasura-bootstrap.mjs
|
||||
- name: dlob-worker-script
|
||||
files:
|
||||
- dlob-worker/worker.mjs
|
||||
- name: dlob-depth-worker-script
|
||||
files:
|
||||
- dlob-depth-worker/worker.mjs
|
||||
- name: dlob-slippage-worker-script
|
||||
files:
|
||||
- dlob-slippage-worker/worker.mjs
|
||||
- name: dlob-ts-archiver-script
|
||||
files:
|
||||
- dlob-ts-archiver/worker.mjs
|
||||
- name: candles-cache-worker-script
|
||||
files:
|
||||
- candles-cache-worker/worker.mjs
|
||||
- name: trade-api-wrapper
|
||||
files:
|
||||
- api/wrapper.mjs
|
||||
- name: trade-api-upstream
|
||||
files:
|
||||
- api/server.mjs
|
||||
|
||||
generatorOptions:
|
||||
disableNameSuffixHash: true
|
||||
|
||||
Reference in New Issue
Block a user