2523 lines
119 KiB
TypeScript
2523 lines
119 KiB
TypeScript
import { config } from '../config/index.js';
|
|
import { IExchangeConnector, ExchangeCapabilities } from '../connectors/types.js';
|
|
import { SignalDirection } from '../strategies/rules/types.js';
|
|
import logger from '../utils/logger.js';
|
|
import * as runtimeOrderRepository from './runtimeOrderRepository.js';
|
|
import { healthTracker } from './healthTracker.js';
|
|
import type { ApiServer } from './apiServer.js';
|
|
import { SymbolMapper } from '../utils/symbolMapper.js';
|
|
import { entryLockService } from './distributedLockService.js';
|
|
import { Notifier } from './notifier.js';
|
|
import {
|
|
normalizeOrderAction,
|
|
normalizeOrderStatus,
|
|
normalizeOrderType,
|
|
normalizeTradeSide
|
|
} from '../domain/tradingEnums.js';
|
|
import { capitalLedger } from './CapitalLedger.js';
|
|
import { randomUUID } from 'crypto';
|
|
import { observabilityService } from './observabilityService.js';
|
|
import {
|
|
AlpacaSubTagIntent,
|
|
buildAlpacaSubTag,
|
|
extractOrderSubTag,
|
|
isBytelystSubTag,
|
|
shouldAttachAlpacaSubTag,
|
|
subTagBelongsToProfile
|
|
} from '../utils/alpacaSubTag.js';
|
|
|
|
const normalizeThrown = (value: unknown): Error => {
|
|
if (value instanceof Error) return value;
|
|
if (value && typeof value === 'object') return new Error(JSON.stringify(value));
|
|
return new Error(String(value ?? 'Unknown error'));
|
|
};
|
|
|
|
const getReconciliationFillQty = (order: any): number => {
|
|
const candidates = [order?.filled_qty, order?.filledQty, order?.filled_quantity, order?.qty, order?.amount, order?.size];
|
|
for (const candidate of candidates) {
|
|
const parsed = Number(candidate);
|
|
if (Number.isFinite(parsed) && parsed > 0) return parsed;
|
|
}
|
|
return 0;
|
|
};
|
|
|
|
const getReconciliationFillPrice = (order: any): number => {
|
|
const candidates = [order?.filled_avg_price, order?.avg_price, order?.price, order?.requestedPrice, order?.requested_price];
|
|
for (const candidate of candidates) {
|
|
const parsed = Number(candidate);
|
|
if (Number.isFinite(parsed) && parsed > 0) return parsed;
|
|
}
|
|
return 0;
|
|
};
|
|
|
|
export interface PositionState {
|
|
symbol: string;
|
|
side: SignalDirection;
|
|
entryPrice: number;
|
|
size: number;
|
|
stopLoss: number;
|
|
takeProfit: number;
|
|
peakPrice: number;
|
|
profitGuardActive?: boolean;
|
|
userId?: string;
|
|
profileId?: string;
|
|
tradeId?: string;
|
|
}
|
|
|
|
export interface PendingOrder {
|
|
orderId: string;
|
|
symbol: string;
|
|
side: SignalDirection;
|
|
qty: number;
|
|
type: 'market' | 'limit';
|
|
requestedPrice: number;
|
|
stopLoss: number;
|
|
takeProfit: number;
|
|
tradeId?: string;
|
|
userId?: string;
|
|
profileId?: string;
|
|
subTag?: string;
|
|
placedAt: number;
|
|
action: 'ENTRY' | 'EXIT';
|
|
reservedAmount?: number;
|
|
}
|
|
|
|
export type ExitLifecycleState =
|
|
| 'idle'
|
|
| 'initiated'
|
|
| 'order_placed'
|
|
| 'verifying'
|
|
| 'filled'
|
|
| 'failed'
|
|
| 'quarantined';
|
|
|
|
export interface ExitLifecycleRecord {
|
|
state: ExitLifecycleState;
|
|
updatedAt: number;
|
|
reason: string;
|
|
orderId?: string;
|
|
details?: string;
|
|
}
|
|
|
|
export interface ExitFillApplyResult {
|
|
success: boolean;
|
|
fullyClosed: boolean;
|
|
appliedQty: number;
|
|
remainingSize: number;
|
|
error?: string;
|
|
}
|
|
|
|
export class TradeExecutor {
|
|
private activeTraders: Map<string, PositionState> = new Map();
|
|
private cooldowns: Map<string, number> = new Map();
|
|
private pendingOrders: Map<string, PendingOrder> = new Map(); // orderId -> PendingOrder
|
|
private exitLifecycle: Map<string, ExitLifecycleRecord> = new Map();
|
|
private entryAutoReduceLastAlertAt: Map<string, number> = new Map();
|
|
private tradeSequence = 0;
|
|
private notifier: Notifier;
|
|
private static readonly POSITION_KEY_SEPARATOR = '::';
|
|
private accountSnapshotTimer?: NodeJS.Timeout;
|
|
private static warnedCapabilities = new Set<string>();
|
|
private profileSettings?: any;
|
|
|
|
constructor(
|
|
private exchange: IExchangeConnector,
|
|
private apiServer?: ApiServer,
|
|
private userId: string = 'global',
|
|
private profileId?: string
|
|
) {
|
|
this.notifier = new Notifier();
|
|
this.startAccountSnapshotPolling();
|
|
}
|
|
|
|
public verifyCapability(capability: keyof ExchangeCapabilities, description: string): boolean {
|
|
const caps = this.exchange.getCapabilities();
|
|
const supported = !!caps[capability];
|
|
if (supported) return true;
|
|
|
|
const capKey = String(capability);
|
|
if (!TradeExecutor.warnedCapabilities.has(capKey)) {
|
|
TradeExecutor.warnedCapabilities.add(capKey);
|
|
logger.warn(`[Executor] Exchange does not support ${description}. Feature will be bypassed.`);
|
|
}
|
|
observabilityService.incrementUnsupportedFeature(capKey);
|
|
return false;
|
|
}
|
|
|
|
public dispose(): void {
|
|
this.clearAccountSnapshotTimer();
|
|
}
|
|
|
|
public setProfileSettings(profileSettings?: any): void {
|
|
this.profileSettings = profileSettings;
|
|
}
|
|
|
|
private clearAccountSnapshotTimer(): void {
|
|
if (!this.accountSnapshotTimer) return;
|
|
clearInterval(this.accountSnapshotTimer);
|
|
this.accountSnapshotTimer = undefined;
|
|
}
|
|
|
|
private startAccountSnapshotPolling(): void {
|
|
if (!this.apiServer) return;
|
|
const fetchAccountSnapshot = this.exchange.fetchAccountSnapshot;
|
|
if (typeof fetchAccountSnapshot !== 'function') return;
|
|
|
|
const refreshSnapshot = async () => {
|
|
try {
|
|
const snapshot = await fetchAccountSnapshot.call(this.exchange);
|
|
if (!snapshot) return;
|
|
this.apiServer!.updateAccountSnapshot({
|
|
...snapshot,
|
|
profileId: this.profileId,
|
|
userId: this.userId
|
|
});
|
|
} catch (error: any) {
|
|
logger.warn(`[Executor] Account snapshot failed for profile ${this.profileId || 'global'}: ${error.message || error}`);
|
|
}
|
|
};
|
|
|
|
refreshSnapshot();
|
|
this.accountSnapshotTimer = setInterval(refreshSnapshot, config.ACCOUNT_SNAPSHOT_INTERVAL_MS);
|
|
if (typeof this.accountSnapshotTimer?.unref === 'function') {
|
|
this.accountSnapshotTimer.unref();
|
|
}
|
|
}
|
|
|
|
private buildPositionKey(symbol: string, tradeId?: string): string {
|
|
const normalizedTradeId = String(tradeId || '').trim();
|
|
if (!normalizedTradeId) return symbol;
|
|
return `${symbol}${TradeExecutor.POSITION_KEY_SEPARATOR}${normalizedTradeId}`;
|
|
}
|
|
|
|
private getLedgerProfileId(candidate?: string): string | undefined {
|
|
const normalized = String(candidate || this.profileId || '').trim();
|
|
if (!normalized || normalized === 'global') return undefined;
|
|
return normalized;
|
|
}
|
|
|
|
private shouldUseAlpacaSubTag(profileScope?: string): boolean {
|
|
const profileId = String(profileScope || this.profileId || '').trim();
|
|
return shouldAttachAlpacaSubTag({
|
|
profileId,
|
|
profileSettings: this.profileSettings
|
|
});
|
|
}
|
|
|
|
private buildOrderSubTag(tradeId: string | undefined, intent: AlpacaSubTagIntent): string | undefined {
|
|
const profileId = String(this.profileId || '').trim();
|
|
if (!profileId) return undefined;
|
|
if (!this.shouldUseAlpacaSubTag(profileId)) return undefined;
|
|
const subTag = buildAlpacaSubTag({
|
|
profileId,
|
|
tradeId,
|
|
intent
|
|
});
|
|
return subTag || undefined;
|
|
}
|
|
|
|
private filterOrdersBySubTagProfile(orders: any[]): any[] {
|
|
const profileId = String(this.profileId || '').trim();
|
|
if (!profileId) return orders;
|
|
if (!this.shouldUseAlpacaSubTag(profileId)) return orders;
|
|
|
|
let filteredOut = 0;
|
|
const scoped = (orders || []).filter((order) => {
|
|
const subTag = extractOrderSubTag(order);
|
|
if (!subTag) return true;
|
|
if (!isBytelystSubTag(subTag)) return true;
|
|
if (subTagBelongsToProfile(subTag, profileId)) return true;
|
|
filteredOut += 1;
|
|
return false;
|
|
});
|
|
|
|
if (filteredOut > 0) {
|
|
logger.info('[Executor] Scoped exchange orders by Alpaca sub-tag', {
|
|
event: 'alpaca_subtag_scope',
|
|
profileId,
|
|
kept: scoped.length,
|
|
dropped: filteredOut
|
|
});
|
|
}
|
|
return scoped;
|
|
}
|
|
|
|
private getPositionsForSymbol(symbol: string): Array<{ key: string; position: PositionState }> {
|
|
const matches: Array<{ key: string; position: PositionState }> = [];
|
|
for (const [key, position] of this.activeTraders.entries()) {
|
|
const keySymbol = key.includes(TradeExecutor.POSITION_KEY_SEPARATOR)
|
|
? key.split(TradeExecutor.POSITION_KEY_SEPARATOR)[0]
|
|
: key;
|
|
if ((position.symbol || keySymbol) === symbol) {
|
|
matches.push({ key, position });
|
|
}
|
|
}
|
|
return matches;
|
|
}
|
|
|
|
private rankPositionCandidate(position: PositionState): number {
|
|
const tradeId = String(position.tradeId || '').trim();
|
|
const hasStableTradeId = tradeId.length > 0 && !tradeId.endsWith('-SYNC');
|
|
return (hasStableTradeId ? 100 : 0) + Math.min(99, Math.round(Number(position.size || 0) * 10));
|
|
}
|
|
|
|
private resolvePositionSelection(symbol: string, tradeId?: string): { key: string; position: PositionState } | null {
|
|
const normalizedTradeId = String(tradeId || '').trim();
|
|
const candidates = this.getPositionsForSymbol(symbol);
|
|
if (candidates.length === 0) return null;
|
|
|
|
if (normalizedTradeId) {
|
|
const exact = candidates.find((entry) => String(entry.position.tradeId || '').trim() === normalizedTradeId);
|
|
if (exact) return exact;
|
|
}
|
|
|
|
candidates.sort((a, b) => {
|
|
const rankDiff = this.rankPositionCandidate(b.position) - this.rankPositionCandidate(a.position);
|
|
if (rankDiff !== 0) return rankDiff;
|
|
const tradeA = String(a.position.tradeId || '');
|
|
const tradeB = String(b.position.tradeId || '');
|
|
return tradeA.localeCompare(tradeB);
|
|
});
|
|
return candidates[0];
|
|
}
|
|
|
|
private upsertPosition(symbol: string, position: PositionState): void {
|
|
const key = this.buildPositionKey(symbol, position.tradeId);
|
|
this.activeTraders.set(key, {
|
|
...position,
|
|
symbol
|
|
});
|
|
}
|
|
|
|
private removePosition(symbol: string, tradeId?: string): void {
|
|
const normalizedTradeId = String(tradeId || '').trim();
|
|
if (normalizedTradeId) {
|
|
const key = this.buildPositionKey(symbol, normalizedTradeId);
|
|
this.activeTraders.delete(key);
|
|
return;
|
|
}
|
|
|
|
for (const [key, position] of this.activeTraders.entries()) {
|
|
if ((position.symbol || symbol) === symbol) {
|
|
this.activeTraders.delete(key);
|
|
}
|
|
}
|
|
}
|
|
|
|
// --- State Accessors ---
|
|
public getActivePosition(symbol: string, tradeId?: string): PositionState | null {
|
|
const selected = this.resolvePositionSelection(symbol, tradeId);
|
|
return selected ? selected.position : null;
|
|
}
|
|
|
|
public getActivePositions(symbol: string): PositionState[] {
|
|
return this.getPositionsForSymbol(symbol)
|
|
.map((entry) => entry.position)
|
|
.sort((a, b) => this.rankPositionCandidate(b) - this.rankPositionCandidate(a));
|
|
}
|
|
|
|
public isSymbolLocked(symbol: string): boolean {
|
|
return this.getPositionsForSymbol(symbol).length > 0;
|
|
}
|
|
|
|
public getActiveSymbols(): string[] {
|
|
return Array.from(new Set(
|
|
Array.from(this.activeTraders.values())
|
|
.map((position) => position.symbol)
|
|
.filter(Boolean)
|
|
));
|
|
}
|
|
|
|
public getAllPositions(): Map<string, PositionState> {
|
|
return this.activeTraders;
|
|
}
|
|
|
|
public getOpenPositionCount(): number {
|
|
return this.activeTraders.size;
|
|
}
|
|
|
|
public getPendingOrders(): Map<string, PendingOrder> {
|
|
return this.pendingOrders;
|
|
}
|
|
|
|
public getExitLifecycle(symbol: string): ExitLifecycleRecord {
|
|
return this.exitLifecycle.get(symbol) || {
|
|
state: 'idle',
|
|
updatedAt: 0,
|
|
reason: 'not_started'
|
|
};
|
|
}
|
|
|
|
public markExitManualReview(symbol: string, reason: string, details?: string, tradeId?: string): void {
|
|
const normalizedReason = String(reason || 'manual_review').trim() || 'manual_review';
|
|
const normalizedDetails = String(details || '').trim() || undefined;
|
|
this.setExitLifecycle(symbol, 'quarantined', normalizedReason, normalizedDetails);
|
|
observabilityService.emitEvent({
|
|
type: 'EXIT_FILL_COHERENCE_VIOLATION',
|
|
severity: 'ERROR',
|
|
message: `Manual review required for ${symbol}: ${normalizedReason}${normalizedDetails ? ` (${normalizedDetails})` : ''}`,
|
|
profileId: this.profileId,
|
|
userId: this.userId,
|
|
symbol
|
|
});
|
|
|
|
const selected = this.resolvePositionSelection(symbol, tradeId);
|
|
const position = selected?.position;
|
|
if (this.apiServer && position) {
|
|
this.apiServer.recordOrderFailure({
|
|
profileId: position.profileId || this.profileId,
|
|
userId: position.userId || this.userId,
|
|
symbol,
|
|
side: position.side === SignalDirection.SELL ? 'BUY' : 'SELL',
|
|
qty: position.size,
|
|
reason: normalizedReason,
|
|
tradeId: String(position.tradeId || tradeId || '').trim() || undefined,
|
|
timestamp: Date.now()
|
|
});
|
|
}
|
|
}
|
|
|
|
public checkCooldown(symbol: string, durationMs: number = 3600000): boolean {
|
|
const lastExit = this.cooldowns.get(symbol) || 0;
|
|
if (Date.now() - lastExit < durationMs) {
|
|
logger.info(`[Executor] 💤 ${symbol} is in cooldown.`);
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
private buildDeterministicTradeId(symbol: string, side: SignalDirection): string {
|
|
const owner = (this.profileId || this.userId || 'global').replace(/[^A-Za-z0-9]/g, '').slice(-12) || 'global';
|
|
const normalizedSymbol = symbol.replace(/[^A-Za-z0-9]/g, '').slice(0, 16) || 'asset';
|
|
this.tradeSequence = (this.tradeSequence + 1) % 1_000_000;
|
|
const sequence = this.tradeSequence.toString().padStart(6, '0');
|
|
return `TRD-${owner}-${normalizedSymbol}-${side}-${Date.now()}-${sequence}`;
|
|
}
|
|
|
|
private buildDeterministicSyncTradeId(symbol: string): string {
|
|
const owner = (this.profileId || this.userId || 'global').replace(/[^A-Za-z0-9]/g, '').slice(-12) || 'global';
|
|
const normalizedSymbol = symbol.replace(/[^A-Za-z0-9]/g, '').slice(0, 16) || 'asset';
|
|
return `TRD-SYNC-${owner}-${normalizedSymbol}`;
|
|
}
|
|
|
|
private hasPendingAction(symbol: string, action: 'ENTRY' | 'EXIT'): boolean {
|
|
for (const pending of this.pendingOrders.values()) {
|
|
if (pending.symbol !== symbol) continue;
|
|
if ((pending.action || '').toUpperCase() === action) {
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
private async hasActiveTradeId(tradeId?: string): Promise<boolean> {
|
|
const normalized = String(tradeId || '').trim();
|
|
if (!normalized) return false;
|
|
return await runtimeOrderRepository.hasActiveOrderForTradeId(normalized, this.profileId);
|
|
}
|
|
|
|
private async isTradeAlreadyFinalized(tradeId?: string): Promise<boolean> {
|
|
const normalized = String(tradeId || '').trim();
|
|
if (!normalized) return false;
|
|
return await runtimeOrderRepository.hasFinalizedTradeHistory(normalized, this.profileId);
|
|
}
|
|
private setExitLifecycle(
|
|
symbol: string,
|
|
state: ExitLifecycleState,
|
|
reason: string,
|
|
details?: string,
|
|
orderId?: string
|
|
): void {
|
|
this.exitLifecycle.set(symbol, {
|
|
state,
|
|
reason,
|
|
details,
|
|
orderId,
|
|
updatedAt: Date.now()
|
|
});
|
|
}
|
|
|
|
private rebuildLifecycleFromPendingOrders(): void {
|
|
for (const pending of this.pendingOrders.values()) {
|
|
if ((pending.action || '').toUpperCase() === 'EXIT') {
|
|
this.setExitLifecycle(
|
|
pending.symbol,
|
|
'order_placed',
|
|
'Restart recovery',
|
|
undefined,
|
|
pending.orderId
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
private normalizeSignalDirection(value?: string | SignalDirection): SignalDirection {
|
|
const upper = String(value || 'BUY').trim().toUpperCase();
|
|
return upper === 'SELL' ? SignalDirection.SELL : SignalDirection.BUY;
|
|
}
|
|
|
|
private normalizeOrderType(value?: string): 'market' | 'limit' {
|
|
const normalized = String(value || 'market').trim().toLowerCase();
|
|
return normalized === 'limit' ? 'limit' : 'market';
|
|
}
|
|
|
|
private roundDownQty(value: number): number {
|
|
if (!Number.isFinite(value) || value <= 0) return 0;
|
|
const precision = Math.max(0, Math.min(10, Math.floor(Number(config.QUANTITY_PRECISION || 6))));
|
|
const factor = Math.pow(10, precision);
|
|
return Math.floor(value * factor) / factor;
|
|
}
|
|
|
|
private resolveOrderReferencePrice(symbol: string, candidatePrice?: number): number {
|
|
const requested = Number(candidatePrice ?? 0);
|
|
if (Number.isFinite(requested) && requested > 0) {
|
|
return requested;
|
|
}
|
|
|
|
const state = this.apiServer?.getState();
|
|
const directPrice = Number(state?.symbols?.[symbol]?.price || 0);
|
|
if (Number.isFinite(directPrice) && directPrice > 0) {
|
|
return directPrice;
|
|
}
|
|
|
|
const dataSymbol = SymbolMapper.toDataSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
if (dataSymbol !== symbol) {
|
|
const mappedPrice = Number(state?.symbols?.[dataSymbol]?.price || 0);
|
|
if (Number.isFinite(mappedPrice) && mappedPrice > 0) {
|
|
return mappedPrice;
|
|
}
|
|
}
|
|
|
|
const minimum = Number(config.MIN_NOTIONAL_USD || 0);
|
|
return Number.isFinite(minimum) && minimum > 0 ? minimum : 0;
|
|
}
|
|
|
|
private isStrictCapitalGuardEnabled(): boolean {
|
|
return config.ENABLE_STRICT_CAPITAL_GUARD !== false;
|
|
}
|
|
|
|
private strictCapitalCostMultiplier(): number {
|
|
if (!this.isStrictCapitalGuardEnabled()) return 1;
|
|
const slippagePct = Math.max(0, Number(config.STRICT_CAPITAL_SLIPPAGE_BUFFER_PCT || 0));
|
|
const feePct = Math.max(0, Number(config.STRICT_CAPITAL_FEE_BUFFER_PCT || 0));
|
|
const multiplier = 1 + ((slippagePct + feePct) / 100);
|
|
return Number.isFinite(multiplier) && multiplier > 0 ? multiplier : 1;
|
|
}
|
|
|
|
private strictCapitalMinReserveUsd(): number {
|
|
if (!this.isStrictCapitalGuardEnabled()) return 0;
|
|
const reserve = Number(config.STRICT_CAPITAL_MIN_RESERVE_USD || 0);
|
|
return Number.isFinite(reserve) && reserve > 0 ? reserve : 0;
|
|
}
|
|
|
|
private clampBuyQtyToAvailableCapital(
|
|
symbol: string,
|
|
requestedQty: number,
|
|
requestedPrice: number | undefined,
|
|
availableCapital: number
|
|
): number {
|
|
if (!Number.isFinite(requestedQty) || requestedQty <= 0) return 0;
|
|
if (!Number.isFinite(availableCapital) || availableCapital <= 0) return 0;
|
|
|
|
const bufferPct = Math.max(0, Number(config.ENTRY_CAPITAL_BUFFER_PCT || 0)) / 100;
|
|
const minimumReserve = this.strictCapitalMinReserveUsd();
|
|
const budget = Math.max(0, (availableCapital * (1 - bufferPct)) - minimumReserve);
|
|
if (!(budget > 0)) return 0;
|
|
|
|
const unitPrice = this.resolveOrderReferencePrice(symbol, requestedPrice);
|
|
if (!(unitPrice > 0)) return 0;
|
|
|
|
const effectiveUnitCost = unitPrice * this.strictCapitalCostMultiplier();
|
|
if (!(effectiveUnitCost > 0)) return 0;
|
|
const maxQty = this.roundDownQty(budget / effectiveUnitCost);
|
|
if (!(maxQty > 0)) return 0;
|
|
return Math.min(requestedQty, maxQty);
|
|
}
|
|
|
|
private maybeEmitEntryAutoReduceAdvisory(params: {
|
|
symbol: string;
|
|
profileId?: string;
|
|
userId?: string;
|
|
requestedQty: number;
|
|
clampedQty: number;
|
|
referencePrice: number;
|
|
availableCapital: number;
|
|
}): void {
|
|
const requestedQty = Number(params.requestedQty || 0);
|
|
const clampedQty = Number(params.clampedQty || 0);
|
|
if (!(requestedQty > 0) || !(clampedQty > 0) || clampedQty >= requestedQty) return;
|
|
|
|
const reducedQty = requestedQty - clampedQty;
|
|
const reductionPct = reducedQty / requestedQty;
|
|
const reductionNotional = reducedQty * Math.max(0, Number(params.referencePrice || 0));
|
|
const minPct = Math.max(0, Number(config.ENTRY_AUTO_REDUCE_ALERT_MIN_PCT || 0));
|
|
const minUsd = Math.max(0, Number(config.ENTRY_AUTO_REDUCE_ALERT_MIN_USD || 0));
|
|
if (reductionPct < minPct && reductionNotional < minUsd) return;
|
|
|
|
const throttleMs = Math.max(0, Number(config.ENTRY_AUTO_REDUCE_ALERT_THROTTLE_MS || 0));
|
|
const profileKey = String(params.profileId || 'global').trim() || 'global';
|
|
const symbolKey = String(params.symbol || '').trim().toUpperCase();
|
|
const throttleKey = `${profileKey}:${symbolKey}`;
|
|
const now = Date.now();
|
|
const lastAt = this.entryAutoReduceLastAlertAt.get(throttleKey) || 0;
|
|
if (throttleMs > 0 && now - lastAt < throttleMs) return;
|
|
this.entryAutoReduceLastAlertAt.set(throttleKey, now);
|
|
|
|
const message = `BUY qty auto-reduced for ${params.symbol}: ${(reductionPct * 100).toFixed(1)}% (~$${reductionNotional.toFixed(2)}) to stay within available capital ($${Number(params.availableCapital || 0).toFixed(2)}). Consider increasing profile capital if this repeats.`;
|
|
observabilityService.emitEvent({
|
|
type: 'INSUFFICIENT_BUYING_POWER',
|
|
severity: 'INFO',
|
|
message,
|
|
profileId: params.profileId,
|
|
userId: params.userId,
|
|
symbol: params.symbol
|
|
});
|
|
}
|
|
|
|
private computeReservedAmount(record: any, fallbackPrice?: number): number {
|
|
const qty = Number(record?.qty ?? record?.quantity ?? record?.amount ?? 0);
|
|
if (!Number.isFinite(qty) || qty <= 0) return 0;
|
|
const candidates = [
|
|
record?.price,
|
|
record?.requested_price,
|
|
record?.requestedPrice,
|
|
record?.filled_avg_price,
|
|
record?.last_price,
|
|
fallbackPrice
|
|
];
|
|
let price = 0;
|
|
for (const candidate of candidates) {
|
|
const numeric = Number(candidate);
|
|
if (Number.isFinite(numeric) && numeric > 0) {
|
|
price = numeric;
|
|
break;
|
|
}
|
|
}
|
|
if (price <= 0) {
|
|
const fallbackNumeric = Number(fallbackPrice ?? 0);
|
|
if (Number.isFinite(fallbackNumeric) && fallbackNumeric > 0) {
|
|
price = fallbackNumeric;
|
|
} else {
|
|
price = config.MIN_NOTIONAL_USD;
|
|
}
|
|
}
|
|
const notional = qty * price;
|
|
const minimum = Number.isFinite(config.MIN_NOTIONAL_USD) && config.MIN_NOTIONAL_USD > 0 ? config.MIN_NOTIONAL_USD : 0;
|
|
return Number(Math.max(notional, minimum));
|
|
}
|
|
|
|
private estimateOrderCost(symbol: string, qty: number, price?: number): number {
|
|
if (!Number.isFinite(qty) || qty <= 0) return 0;
|
|
const unitPrice = this.resolveOrderReferencePrice(symbol, price);
|
|
const notional = qty * unitPrice * this.strictCapitalCostMultiplier();
|
|
const minimum = Number.isFinite(config.MIN_NOTIONAL_USD) && config.MIN_NOTIONAL_USD > 0 ? config.MIN_NOTIONAL_USD : 0;
|
|
const base = Math.max(notional, minimum);
|
|
return Number(base + this.strictCapitalMinReserveUsd());
|
|
}
|
|
|
|
private buildPendingOrderFromRow(row: any): PendingOrder | null {
|
|
const orderId = String(row?.id || row?.order_id || '').trim();
|
|
if (!orderId) return null;
|
|
const symbol = String(row?.symbol || '').trim();
|
|
if (!symbol) return null;
|
|
const side = this.normalizeSignalDirection(row?.side);
|
|
const qty = Number(row?.qty || row?.quantity || row?.filled_qty || 0);
|
|
if (!Number.isFinite(qty) || qty <= 0) return null;
|
|
const createdAt = row?.timestamp
|
|
? Number(row.timestamp)
|
|
: row?.created_at
|
|
? Number(Date.parse(row.created_at))
|
|
: Date.now();
|
|
|
|
const requestedPrice = Number(row?.price || row?.requested_price || row?.requestedPrice || 0);
|
|
const reservedAmount = this.computeReservedAmount(row, requestedPrice);
|
|
const profileId = String(row?.profile_id || this.profileId || '').trim() || undefined;
|
|
|
|
return {
|
|
orderId,
|
|
symbol,
|
|
side,
|
|
qty,
|
|
type: this.normalizeOrderType(row?.type),
|
|
requestedPrice,
|
|
stopLoss: Number(row?.stop_loss || 0),
|
|
takeProfit: Number(row?.take_profit || 0),
|
|
tradeId: String(row?.trade_id || '').trim() || undefined,
|
|
userId: String(row?.user_id || this.userId || '').trim() || undefined,
|
|
profileId,
|
|
subTag: extractOrderSubTag(row) || undefined,
|
|
placedAt: Number.isFinite(createdAt) ? Number(createdAt) : Date.now(),
|
|
action: String(row?.action || (side === SignalDirection.SELL ? 'EXIT' : 'ENTRY')).toUpperCase() as 'ENTRY' | 'EXIT',
|
|
reservedAmount
|
|
};
|
|
}
|
|
|
|
private addPendingOrderFromRecord(record: any): void {
|
|
const pending = this.buildPendingOrderFromRow(record);
|
|
if (!pending) return;
|
|
if (this.pendingOrders.has(pending.orderId)) return;
|
|
this.pendingOrders.set(pending.orderId, pending);
|
|
}
|
|
|
|
private async populatePendingOrdersFromDb(): Promise<void> {
|
|
this.pendingOrders.clear();
|
|
const rows = await runtimeOrderRepository.getOpenOrdersForProfile(this.profileId || '');
|
|
rows.forEach((row) => this.addPendingOrderFromRecord(row));
|
|
}
|
|
|
|
public async fetchExchangeOpenOrders(): Promise<any[]> {
|
|
if (!this.verifyCapability('fetchOpenOrders', 'fetching open orders')) return [];
|
|
try {
|
|
const orders = await this.instrumentExchangeCall('fetch_open_orders', () => this.exchange.fetchOpenOrders!(config.SYMBOLS));
|
|
return this.filterOrdersBySubTagProfile(orders || []);
|
|
} catch (error: any) {
|
|
logger.error(`[Executor] Failed to fetch open orders from exchange: ${error.message || error}`);
|
|
return [];
|
|
}
|
|
}
|
|
|
|
public async fetchExchangeClosedOrders(
|
|
symbols?: string[],
|
|
lookbackHours?: number,
|
|
options?: {
|
|
limitPerPage?: number;
|
|
maxPages?: number;
|
|
}
|
|
): Promise<any[]> {
|
|
if (!this.verifyCapability('fetchClosedOrders', 'fetching closed orders')) return [];
|
|
try {
|
|
const safeLookbackHours = Number.isFinite(Number(lookbackHours))
|
|
? Math.max(1, Math.floor(Number(lookbackHours)))
|
|
: Math.max(1, Math.floor(Number(config.RECON_EXIT_BACKFILL_LOOKBACK_HOURS || 72)));
|
|
const limitPerPage = Math.max(
|
|
1,
|
|
Math.min(500, Math.floor(Number(options?.limitPerPage || config.RECON_ORDER_COVERAGE_FETCH_LIMIT_PER_PAGE || 500)))
|
|
);
|
|
const maxPages = Math.max(
|
|
1,
|
|
Math.min(100, Math.floor(Number(options?.maxPages || config.RECON_ORDER_COVERAGE_MAX_FETCH_PAGES || 8)))
|
|
);
|
|
const after = new Date(Date.now() - safeLookbackHours * 60 * 60 * 1000);
|
|
const targetSymbols = (symbols && symbols.length > 0) ? symbols : config.SYMBOLS;
|
|
const orders = await this.instrumentExchangeCall('fetch_closed_orders', () => this.exchange.fetchClosedOrders!(targetSymbols, {
|
|
after,
|
|
limit: limitPerPage,
|
|
maxPages
|
|
}));
|
|
return this.filterOrdersBySubTagProfile(orders || []);
|
|
} catch (error: any) {
|
|
logger.error(`[Executor] Failed to fetch closed orders from exchange: ${error.message || error}`);
|
|
return [];
|
|
}
|
|
}
|
|
|
|
public async fetchExchangePosition(symbol: string): Promise<any | null> {
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
try {
|
|
return await this.instrumentExchangeCall('get_position', () => this.exchange.getPosition(tradeSymbol));
|
|
} catch (error: any) {
|
|
logger.error(`[Executor] Failed to fetch exchange position for ${symbol}: ${error.message || error}`);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private async populatePendingOrdersFromExchange(): Promise<void> {
|
|
const symbols = config.SYMBOLS;
|
|
const orders = await this.fetchExchangeOpenOrders();
|
|
if (!orders || orders.length === 0) return;
|
|
|
|
orders.forEach((order) => {
|
|
const data = {
|
|
id: order.id || order.client_order_id || order.orderId,
|
|
symbol: order.symbol,
|
|
side: this.normalizeSignalDirection(order.side),
|
|
qty: order.amount || order.qty || order.size,
|
|
type: order.type || order.order_type,
|
|
price: order.price || order.requestedPrice || 0,
|
|
stop_loss: order.stop_loss,
|
|
take_profit: order.take_profit,
|
|
trade_id: order.trade_id,
|
|
user_id: this.userId,
|
|
profile_id: this.profileId,
|
|
sub_tag: extractOrderSubTag(order),
|
|
timestamp: order.timestamp,
|
|
created_at: order.datetime,
|
|
action: this.normalizeSignalDirection(order.side) === SignalDirection.SELL ? 'EXIT' : 'ENTRY'
|
|
};
|
|
if (!symbols.some(sym => String(sym).toUpperCase() === String(data.symbol).toUpperCase())) {
|
|
data.symbol = SymbolMapper.toDataSymbol(String(order.symbol || ''), config.EXECUTION_PROVIDER);
|
|
}
|
|
this.addPendingOrderFromRecord(data);
|
|
});
|
|
}
|
|
|
|
public async rebuildStartupState(): Promise<void> {
|
|
await this.populatePendingOrdersFromDb();
|
|
await this.populatePendingOrdersFromExchange();
|
|
this.rebuildLifecycleFromPendingOrders();
|
|
await this.rebuildCapitalLedgerFromState();
|
|
}
|
|
|
|
private async rebuildCapitalLedgerFromState(): Promise<void> {
|
|
const ledgerProfileId = this.getLedgerProfileId();
|
|
if (!ledgerProfileId) return;
|
|
try {
|
|
const reservedOrders = Array.from(this.pendingOrders.values())
|
|
.filter((pending) => pending.profileId === ledgerProfileId && pending.action === 'ENTRY')
|
|
.reduce((sum, pending) => {
|
|
const amount = pending.reservedAmount ?? this.computeReservedAmount(pending, pending.requestedPrice);
|
|
return sum + amount;
|
|
}, 0);
|
|
|
|
let reservedPositions = 0;
|
|
for (const symbol of config.SYMBOLS) {
|
|
const virtualPosition = await runtimeOrderRepository.getVirtualOpenPosition(ledgerProfileId, symbol);
|
|
if (virtualPosition && virtualPosition.qty > 0 && virtualPosition.entryPrice > 0) {
|
|
reservedPositions += virtualPosition.qty * virtualPosition.entryPrice;
|
|
}
|
|
}
|
|
|
|
await capitalLedger.rebuildLedger(ledgerProfileId, reservedOrders, reservedPositions);
|
|
} catch (error: any) {
|
|
logger.warn(`[Executor] Failed to rebuild capital ledger for ${ledgerProfileId}: ${error.message}`);
|
|
}
|
|
}
|
|
|
|
private resolveCapitalLedgerDriftScope(): 'exchange' | 'virtual' {
|
|
const configured = String(config.CAPITAL_LEDGER_DRIFT_SCOPE || 'auto').trim().toLowerCase();
|
|
if (configured === 'exchange' || configured === 'virtual') return configured;
|
|
|
|
// In shared-account profile mode, exchange position notional is account-level and
|
|
// cannot be attributed safely per profile. Default to profile-scoped virtual notional.
|
|
const normalizedProfileId = String(this.profileId || '').trim().toLowerCase();
|
|
const isDedicatedProfileScope = normalizedProfileId.length > 0
|
|
&& normalizedProfileId !== 'global'
|
|
&& !normalizedProfileId.startsWith('default-');
|
|
return isDedicatedProfileScope ? 'virtual' : 'exchange';
|
|
}
|
|
|
|
private async computeExchangeOpenNotional(symbols: string[]): Promise<number> {
|
|
let exchangeNotional = 0;
|
|
for (const symbol of symbols) {
|
|
try {
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
const pos = await this.instrumentExchangeCall('get_position', () => this.exchange.getPosition(tradeSymbol));
|
|
const qty = Math.abs(Number(pos?.qty || 0));
|
|
const price = Number(pos?.avg_entry_price || pos?.current_price || 0);
|
|
if (qty > 0 && price > 0) exchangeNotional += qty * price;
|
|
} catch (_) {
|
|
// per-symbol error; continue
|
|
}
|
|
}
|
|
return exchangeNotional;
|
|
}
|
|
|
|
private async computeProfileVirtualNotional(profileId: string, symbols: string[]): Promise<number> {
|
|
let virtualNotional = 0;
|
|
for (const symbol of symbols) {
|
|
try {
|
|
const virtualPosition = await runtimeOrderRepository.getVirtualOpenPosition(profileId, symbol);
|
|
if (virtualPosition && virtualPosition.qty > 0 && virtualPosition.entryPrice > 0) {
|
|
virtualNotional += virtualPosition.qty * virtualPosition.entryPrice;
|
|
}
|
|
} catch (_) {
|
|
// per-symbol error; continue
|
|
}
|
|
}
|
|
return virtualNotional;
|
|
}
|
|
|
|
/**
|
|
* FIX-03: Cross-validates the capital ledger's reserved_for_positions against
|
|
* the actual open position notional on the exchange. Emits CAPITAL_LEDGER_DRIFT
|
|
* if the discrepancy exceeds the configured threshold.
|
|
*/
|
|
public async crossValidateCapitalLedger(symbols: string[]): Promise<void> {
|
|
const ledgerProfileId = this.getLedgerProfileId();
|
|
if (!ledgerProfileId) return;
|
|
try {
|
|
const scope = this.resolveCapitalLedgerDriftScope();
|
|
const observedNotional = scope === 'exchange'
|
|
? await this.computeExchangeOpenNotional(symbols)
|
|
: await this.computeProfileVirtualNotional(ledgerProfileId, symbols);
|
|
|
|
const ledger = await capitalLedger.getLedger(ledgerProfileId);
|
|
const ledgerReserved = Number(ledger?.reserved_for_positions || 0);
|
|
const delta = Math.abs(observedNotional - ledgerReserved);
|
|
const driftThresholdPct = Math.max(1, Number(config.CAPITAL_LEDGER_DRIFT_ALERT_PCT || 10)) / 100;
|
|
const minDriftUsd = Math.max(0, Number(config.CAPITAL_LEDGER_DRIFT_MIN_USD || 10));
|
|
const maxAllowed = Math.max(observedNotional, ledgerReserved) * driftThresholdPct;
|
|
|
|
if (delta > maxAllowed && delta > minDriftUsd) {
|
|
logger.error(`[Executor] ⚠️ CAPITAL_LEDGER_DRIFT: scope=${scope}, observed notional=${observedNotional.toFixed(2)}, ledger reserved=${ledgerReserved.toFixed(2)}, delta=${delta.toFixed(2)} | Profile: ${ledgerProfileId}`);
|
|
observabilityService.emitEvent({
|
|
type: 'CAPITAL_LEDGER_DRIFT',
|
|
severity: 'ERROR',
|
|
message: `Capital ledger drift at startup (${scope} scope): observed notional $${observedNotional.toFixed(2)} vs ledger reserved $${ledgerReserved.toFixed(2)} (delta $${delta.toFixed(2)}).`,
|
|
profileId: ledgerProfileId
|
|
});
|
|
} else {
|
|
logger.info(`[Executor] ✅ Capital ledger validated: scope=${scope}, observed=${observedNotional.toFixed(2)}, ledger_reserved=${ledgerReserved.toFixed(2)}, delta=${delta.toFixed(2)} | Profile: ${ledgerProfileId}`);
|
|
}
|
|
} catch (e) {
|
|
logger.warn(`[Executor] Capital ledger cross-validation failed for ${ledgerProfileId}: ${e}`);
|
|
}
|
|
}
|
|
|
|
private getFilledQuantity(order: any): number | undefined {
|
|
const candidates = [
|
|
order?.filled_qty,
|
|
order?.filledQty,
|
|
order?.filled_quantity,
|
|
order?.qty_filled,
|
|
order?.executed_qty,
|
|
order?.filled
|
|
];
|
|
|
|
for (const raw of candidates) {
|
|
const value = Number(raw);
|
|
if (Number.isFinite(value) && value > 0) {
|
|
return value;
|
|
}
|
|
}
|
|
|
|
return undefined;
|
|
}
|
|
|
|
private isDuplicateOrderError(error: any): boolean {
|
|
if (!error) return false;
|
|
|
|
const message = String(error?.message || '').toLowerCase();
|
|
const responseData = String(error?.response?.data || error?.data || '').toLowerCase();
|
|
const code = String(error?.code || '').toLowerCase();
|
|
if (message.includes('duplicate') || responseData.includes('duplicate') || code.includes('duplicate')) {
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
// --- Core Execution ---
|
|
|
|
/**
|
|
* Places a direct order to open a position.
|
|
* Does NOT check risk or strategy rules. Assumes caller has validated.
|
|
*/
|
|
public async openPosition(
|
|
symbol: string,
|
|
side: SignalDirection,
|
|
qty: number,
|
|
type: 'market' | 'limit' = 'market',
|
|
price?: number,
|
|
sl?: number,
|
|
tp?: number,
|
|
userIdOverride?: string
|
|
): Promise<{ success: boolean, orderId?: string, error?: string }> {
|
|
if (healthTracker.isPaused()) {
|
|
logger.info(`[TradeExecutor] 🛑 Entry BLOCKED for ${symbol}: Bot is PAUSED by admin.`);
|
|
return { success: false, error: 'Trade execution is paused by administrator' };
|
|
}
|
|
console.log('[TradeExecutor] openPosition called', { symbol, side, qty, type, price, userIdOverride });
|
|
let executionQty = this.roundDownQty(Number(qty));
|
|
if (!(executionQty > 0)) {
|
|
return { success: false, error: 'Invalid order quantity' };
|
|
}
|
|
const ledgerProfileId = this.getLedgerProfileId();
|
|
const tradeId = this.buildDeterministicTradeId(symbol, side);
|
|
const finalUserId = userIdOverride || this.userId;
|
|
let reservedEstimate = ledgerProfileId ? this.estimateOrderCost(symbol, executionQty, price) : 0;
|
|
let activeOrderId: string | undefined;
|
|
let pendingCaptured = false;
|
|
let capitalReserved = false;
|
|
let capitalReservationAmount = reservedEstimate;
|
|
const normalizedSymbol = String(symbol || '').trim();
|
|
let lockAcquired = false;
|
|
let lockOwner: string | undefined;
|
|
let orderSubTag: string | undefined;
|
|
|
|
const releaseCapitalOnAbort = async () => {
|
|
if (capitalReserved && ledgerProfileId && capitalReservationAmount > 0) {
|
|
const amountToRelease = capitalReservationAmount;
|
|
capitalReservationAmount = 0;
|
|
capitalReserved = false;
|
|
await this.releaseLedgerReservation(ledgerProfileId, amountToRelease);
|
|
}
|
|
};
|
|
|
|
const releaseLockIfHeld = async () => {
|
|
if (lockAcquired && ledgerProfileId && lockOwner) {
|
|
lockAcquired = false;
|
|
const released = await entryLockService.releaseRowLock(ledgerProfileId, normalizedSymbol, lockOwner);
|
|
lockOwner = undefined;
|
|
if (!released) {
|
|
logger.warn(`[DistributedLock] Failed to release lock for ${symbol} (${ledgerProfileId})`);
|
|
}
|
|
}
|
|
};
|
|
try {
|
|
if (ledgerProfileId) {
|
|
const ownerCandidate = `${process.pid}-${randomUUID()}`;
|
|
const acquisition = await entryLockService.tryAcquireRowLock(ledgerProfileId, normalizedSymbol, ownerCandidate, 30);
|
|
if (!acquisition) {
|
|
logger.warn(`[EntryLock] Entry locked for ${symbol} (profile=${ledgerProfileId})`);
|
|
return { success: false, error: 'Entry already running for this profile and symbol' };
|
|
}
|
|
lockOwner = ownerCandidate;
|
|
lockAcquired = true;
|
|
const lifecycleActive = await entryLockService.isEntryInProgress(ledgerProfileId, normalizedSymbol);
|
|
if (lifecycleActive) {
|
|
await releaseLockIfHeld();
|
|
logger.warn(`[Executor] Entry lifecycle already open for ${symbol} (profile=${ledgerProfileId})`);
|
|
return { success: false, error: 'Entry lifecycle already exists' };
|
|
}
|
|
if (await this.hasActiveTradeId(tradeId)) {
|
|
await releaseLockIfHeld();
|
|
logger.warn(`[Executor] Duplicate ENTRY request blocked for ${symbol}`);
|
|
return { success: false, error: 'Duplicate entry request blocked (idempotency window)' };
|
|
}
|
|
const available = await capitalLedger.getAvailableCapital(ledgerProfileId);
|
|
const numericAvailable = Number.isFinite(Number(available ?? 0)) ? Number(available ?? 0) : 0;
|
|
if (side === SignalDirection.BUY) {
|
|
const requestedBeforeClamp = executionQty;
|
|
const clampedQty = this.clampBuyQtyToAvailableCapital(symbol, executionQty, price, numericAvailable);
|
|
if (!(clampedQty > 0)) {
|
|
await releaseLockIfHeld();
|
|
logger.warn(`[Executor] Entry blocked for ${symbol}: unable to size BUY within available capital (${numericAvailable}).`);
|
|
return { success: false, error: 'Insufficient capital after execution safety buffer' };
|
|
}
|
|
if (clampedQty + 1e-10 < executionQty) {
|
|
logger.warn(`[Executor] Entry qty clamped for ${symbol}: requested=${executionQty}, clamped=${clampedQty}, available=${numericAvailable}`);
|
|
this.maybeEmitEntryAutoReduceAdvisory({
|
|
symbol,
|
|
profileId: ledgerProfileId,
|
|
userId: finalUserId,
|
|
requestedQty: requestedBeforeClamp,
|
|
clampedQty,
|
|
referencePrice: this.resolveOrderReferencePrice(symbol, price),
|
|
availableCapital: numericAvailable
|
|
});
|
|
executionQty = clampedQty;
|
|
}
|
|
}
|
|
reservedEstimate = this.estimateOrderCost(symbol, executionQty, price);
|
|
capitalReservationAmount = reservedEstimate;
|
|
if (numericAvailable < reservedEstimate) {
|
|
await releaseLockIfHeld();
|
|
logger.warn(`[Executor] Insufficient capital for ${symbol}: available=${numericAvailable}, required=${reservedEstimate}`);
|
|
return { success: false, error: 'Insufficient capital to reserve order' };
|
|
}
|
|
if (!(await capitalLedger.reserveForOrder(ledgerProfileId, reservedEstimate))) {
|
|
await releaseLockIfHeld();
|
|
return { success: false, error: 'Insufficient capital to reserve order' };
|
|
}
|
|
capitalReserved = true;
|
|
capitalReservationAmount = reservedEstimate;
|
|
}
|
|
if (this.hasPendingAction(symbol, 'ENTRY')) {
|
|
logger.warn(`[Executor] Duplicate ENTRY request blocked for ${symbol}`);
|
|
await releaseCapitalOnAbort();
|
|
await releaseLockIfHeld();
|
|
return { success: false, error: 'Duplicate entry request blocked (pending action)' };
|
|
}
|
|
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
const clientOrderId = `bytelyst-${ledgerProfileId || this.profileId || 'global'}-${tradeId}`;
|
|
orderSubTag = this.buildOrderSubTag(tradeId, 'ENTRY');
|
|
if (orderSubTag) {
|
|
logger.info('[Executor] Alpaca sub-tag prepared', {
|
|
event: 'alpaca_subtag_prepared',
|
|
profileId: ledgerProfileId || this.profileId,
|
|
userId: finalUserId,
|
|
symbol,
|
|
tradeId,
|
|
intent: 'ENTRY',
|
|
subTag: orderSubTag
|
|
});
|
|
}
|
|
|
|
// --- Pre-flight BUY Feasibility Check ---
|
|
if (side === SignalDirection.BUY) {
|
|
const snapshot = this.apiServer?.getState()?.accountSnapshot;
|
|
if (snapshot) {
|
|
const buyingPower = Number(snapshot.buying_power || 0);
|
|
if (buyingPower < reservedEstimate) {
|
|
logger.warn(`[Guardrail] Aborting BUY for ${symbol}: Insufficient Broker Buying Power (${buyingPower} < ${reservedEstimate})`);
|
|
if (this.apiServer) {
|
|
this.apiServer.recordOrderFailure({
|
|
profileId: ledgerProfileId,
|
|
userId: finalUserId,
|
|
symbol,
|
|
side: 'BUY',
|
|
qty: executionQty,
|
|
reason: 'INSUFFICIENT_BUYING_POWER',
|
|
tradeId,
|
|
subTag: orderSubTag,
|
|
timestamp: Date.now()
|
|
});
|
|
}
|
|
observabilityService.emitEvent({
|
|
type: 'INSUFFICIENT_BUYING_POWER',
|
|
severity: 'WARN',
|
|
message: `Insufficient Broker Buying Power for ${symbol} ($${reservedEstimate.toFixed(2)} required)`,
|
|
profileId: ledgerProfileId,
|
|
userId: finalUserId,
|
|
symbol
|
|
});
|
|
await releaseCapitalOnAbort();
|
|
await releaseLockIfHeld();
|
|
return { success: false, error: 'INSUFFICIENT_BUYING_POWER' };
|
|
}
|
|
}
|
|
}
|
|
|
|
if (side === SignalDirection.SELL) {
|
|
if (!this.verifyCapability('shorting', 'Short Selling')) {
|
|
await releaseCapitalOnAbort();
|
|
await releaseLockIfHeld();
|
|
return { success: false, error: 'EXCHANGE_DOES_NOT_SUPPORT_SHORTING' };
|
|
}
|
|
}
|
|
|
|
if (await this.isTradeAlreadyFinalized(tradeId)) {
|
|
logger.warn(`[Executor] ENTRY ${tradeId} already finalized; skipping duplicate request for ${symbol}`);
|
|
await releaseCapitalOnAbort();
|
|
await releaseLockIfHeld();
|
|
return { success: false, error: 'Trade lifecycle already finalized' };
|
|
}
|
|
|
|
logger.info(`[Executor] 🚀 Placing ${side.toUpperCase()} ${type} for ${symbol} | Qty: ${executionQty} ${price ? `@ ${price}` : ''} | Trade: ${tradeId}`);
|
|
logger.info('ENTRY submitted', {
|
|
event: 'entry_submitted',
|
|
symbol,
|
|
tradeId,
|
|
profileId: ledgerProfileId,
|
|
userId: finalUserId,
|
|
qty: executionQty,
|
|
price: price || 0,
|
|
side,
|
|
action: 'ENTRY',
|
|
});
|
|
|
|
let order: any;
|
|
try {
|
|
order = await this.instrumentExchangeCall('place_order', () => this.exchange.placeOrder(
|
|
tradeSymbol,
|
|
side.toLowerCase() as 'buy' | 'sell',
|
|
executionQty,
|
|
type,
|
|
price,
|
|
sl,
|
|
tp,
|
|
clientOrderId,
|
|
{
|
|
subTag: orderSubTag,
|
|
profileId: ledgerProfileId || this.profileId,
|
|
tradeId,
|
|
intent: 'ENTRY'
|
|
}
|
|
));
|
|
} catch (error: any) {
|
|
if (this.isDuplicateOrderError(error)) {
|
|
const existingOrderRow = await runtimeOrderRepository.getOrderByTradeId(tradeId, ledgerProfileId);
|
|
const existingOrderId = String(existingOrderRow?.order_id || '').trim();
|
|
if (!existingOrderId) {
|
|
throw normalizeThrown(error);
|
|
}
|
|
if (this.exchange.getOrder) {
|
|
order = await this.exchange.getOrder(existingOrderId, tradeSymbol);
|
|
}
|
|
if (!order) {
|
|
order = {
|
|
id: existingOrderId,
|
|
status: existingOrderRow?.status || 'pending_new',
|
|
filled_avg_price: existingOrderRow?.price || 0,
|
|
filled_qty: existingOrderRow?.qty || 0
|
|
};
|
|
}
|
|
} else {
|
|
const errorMsg = error.message || String(error);
|
|
observabilityService.emitEvent({
|
|
type: 'ORDER_FAILURE',
|
|
severity: 'ERROR',
|
|
message: `Exchange rejected ${side} order for ${symbol}: ${errorMsg}`,
|
|
profileId: ledgerProfileId,
|
|
userId: finalUserId,
|
|
symbol
|
|
});
|
|
throw normalizeThrown(error);
|
|
}
|
|
}
|
|
|
|
if (!order) {
|
|
await releaseCapitalOnAbort();
|
|
await releaseLockIfHeld();
|
|
return { success: false, error: "Order not returned from exchange" };
|
|
}
|
|
|
|
order.client_order_id = order.client_order_id || clientOrderId;
|
|
if (orderSubTag) {
|
|
order.subtag = order.subtag || orderSubTag;
|
|
order.sub_tag = order.sub_tag || orderSubTag;
|
|
}
|
|
|
|
// Log the order immediately (status may be pending)
|
|
const initialPrice = price || order.filled_avg_price || 0;
|
|
await this.logOrderToDb(order, symbol, side, executionQty, initialPrice, type, sl, tp, finalUserId, tradeId, 'ENTRY');
|
|
this.updateDashboardOrder(order, symbol, side, executionQty, initialPrice, type, tradeId, 'ENTRY');
|
|
|
|
// --- Track for background sync ---
|
|
const pendingReservationAmount = capitalReservationAmount;
|
|
this.pendingOrders.set(order.id, {
|
|
orderId: order.id,
|
|
symbol,
|
|
side,
|
|
qty: executionQty,
|
|
type,
|
|
requestedPrice: initialPrice,
|
|
stopLoss: sl || 0,
|
|
takeProfit: tp || 0,
|
|
tradeId,
|
|
userId: finalUserId,
|
|
profileId: ledgerProfileId,
|
|
subTag: orderSubTag,
|
|
reservedAmount: pendingReservationAmount,
|
|
placedAt: Date.now(),
|
|
action: 'ENTRY'
|
|
});
|
|
pendingCaptured = true;
|
|
activeOrderId = order.id;
|
|
await releaseLockIfHeld();
|
|
capitalReserved = false;
|
|
capitalReservationAmount = 0;
|
|
|
|
// --- Order Fill Verification ---
|
|
const verifiedOrder = await this.waitForFill(order.id, symbol, type);
|
|
|
|
const verifiedStatus = (verifiedOrder?.status || 'filled').toLowerCase();
|
|
const terminalStatuses = new Set(['canceled', 'expired', 'rejected', 'unknown']);
|
|
if (!verifiedOrder || terminalStatuses.has(verifiedStatus)) {
|
|
logger.warn(`[Executor] âš ï¸ Order ${order.id} was ${verifiedOrder?.status || 'lost'}. Not tracking position.`);
|
|
// Update order status in DB
|
|
await runtimeOrderRepository.updateOrderStatus(order.id, verifiedOrder?.status || 'canceled');
|
|
const pending = this.pendingOrders.get(order.id);
|
|
await this.releasePendingOrderReservation(pending);
|
|
this.pendingOrders.delete(order.id);
|
|
if (this.apiServer && (verifiedStatus === 'rejected' || verifiedStatus === 'canceled')) {
|
|
this.apiServer.recordOrderFailure({
|
|
profileId: this.profileId,
|
|
userId: this.userId,
|
|
symbol,
|
|
side: side === SignalDirection.SELL ? 'SELL' : 'BUY',
|
|
qty: executionQty,
|
|
reason: `Order ${verifiedStatus}: ${verifiedOrder?.fail_reason || verifiedOrder?.reason || 'no reason provided'}`,
|
|
tradeId,
|
|
subTag: orderSubTag,
|
|
timestamp: Date.now()
|
|
});
|
|
}
|
|
return { success: false, error: `Order ${verifiedOrder?.status || 'not filled'}` };
|
|
}
|
|
|
|
const fillPrice = Number(verifiedOrder?.filled_avg_price || initialPrice || 0);
|
|
const filledQty = this.getFilledQuantity(verifiedOrder) || executionQty;
|
|
|
|
// ✅ Update order status in DB when filled
|
|
await runtimeOrderRepository.updateOrderStatus(
|
|
order.id,
|
|
verifiedStatus,
|
|
new Date(),
|
|
fillPrice > 0 ? fillPrice : undefined,
|
|
filledQty
|
|
);
|
|
|
|
// --- Slippage Guard (for market orders) ---
|
|
if (type === 'market' && price && price > 0 && fillPrice > 0) {
|
|
const slippagePercent = Math.abs((fillPrice - price) / price) * 100;
|
|
if (slippagePercent > config.MAX_SLIPPAGE_PERCENT) {
|
|
logger.warn(`[Executor] âš ï¸ SLIPPAGE WARNING: ${symbol} requested ${price}, filled at ${fillPrice} (${slippagePercent.toFixed(2)}% slippage, max ${config.MAX_SLIPPAGE_PERCENT}%)`);
|
|
await this.notifier.sendAlert(
|
|
`âš ï¸ **SLIPPAGE WARNING**\n${symbol}: ${slippagePercent.toFixed(2)}% slippage\nRequested: $${price}\nFilled: $${fillPrice}`);
|
|
observabilityService.emitEvent({
|
|
type: 'SYSTEM_ERROR',
|
|
severity: 'ERROR',
|
|
message: `Slippage breach for ${symbol}: ${slippagePercent.toFixed(2)}% (max ${config.MAX_SLIPPAGE_PERCENT}%).`,
|
|
profileId: this.profileId,
|
|
userId: finalUserId,
|
|
symbol
|
|
});
|
|
if (config.ENABLE_AUTO_PAUSE_ON_SLIPPAGE_BREACH && !healthTracker.isPaused()) {
|
|
const pauseReason = `Auto-paused by slippage guard: ${symbol} breached max slippage (${slippagePercent.toFixed(2)}% > ${config.MAX_SLIPPAGE_PERCENT}%).`;
|
|
healthTracker.recordTradingControl({
|
|
mode: 'PAUSED',
|
|
lastChangedBy: 'system:auto_slippage_guard',
|
|
lastChangedAt: Date.now(),
|
|
reason: pauseReason
|
|
});
|
|
this.apiServer?.publishHealthSnapshot({ broadcast: true, force: true });
|
|
logger.error(`[Guardrail] ${pauseReason}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Track verified position
|
|
this.upsertPosition(symbol, {
|
|
symbol,
|
|
side,
|
|
entryPrice: fillPrice,
|
|
size: filledQty,
|
|
stopLoss: sl || 0,
|
|
takeProfit: tp || 0,
|
|
peakPrice: fillPrice,
|
|
userId: finalUserId,
|
|
profileId: this.profileId,
|
|
tradeId
|
|
});
|
|
|
|
const pending = this.pendingOrders.get(order.id);
|
|
await this.finalizeEntryReservation(pending, fillPrice, filledQty);
|
|
this.pendingOrders.delete(order.id);
|
|
|
|
logger.info(`[Executor] ✅ Order FILLED: ${symbol} ${side} ${filledQty} @ $${fillPrice} (Trade: ${tradeId})`);
|
|
logger.info('ENTRY filled', {
|
|
event: 'entry_filled',
|
|
symbol,
|
|
tradeId,
|
|
profileId: this.profileId,
|
|
qty: filledQty,
|
|
price: fillPrice,
|
|
userId: finalUserId,
|
|
side,
|
|
});
|
|
await this.notifier.sendAlert(`🚀 **ORDER FILLED**\nSymbol: ${symbol}\nSide: ${side}\nQty: ${filledQty}\nPrice: $${fillPrice}`);
|
|
if (this.apiServer) {
|
|
const allPos = this.getAllActivePositions();
|
|
this.apiServer.updatePositions(allPos, this.profileId || 'global');
|
|
}
|
|
return { success: true, orderId: order.id };
|
|
|
|
} catch (error: any) {
|
|
await releaseCapitalOnAbort();
|
|
await releaseLockIfHeld();
|
|
console.error('[TradeExecutor] openPosition threw', {
|
|
symbol,
|
|
side,
|
|
type,
|
|
qty: executionQty,
|
|
price,
|
|
error: String(error),
|
|
stack: error?.stack
|
|
});
|
|
logger.error('[TradeExecutor] openPosition exception', {
|
|
symbol,
|
|
side,
|
|
type,
|
|
qty: executionQty,
|
|
price,
|
|
error: String(error),
|
|
stack: error?.stack
|
|
});
|
|
logger.error(`[Executor] ⌠Open Failed: ${error.message}`);
|
|
await this.notifier.sendAlert(`⌠**OPEN FAILED**\nSymbol: ${symbol}\nError: ${error.message}`);
|
|
if (this.apiServer) {
|
|
const normalizedSide = side === SignalDirection.SELL ? 'SELL' : 'BUY';
|
|
this.apiServer.recordOrderFailure({
|
|
profileId: this.profileId,
|
|
userId: this.userId,
|
|
symbol,
|
|
side: normalizedSide,
|
|
qty: executionQty,
|
|
reason: error.message ? error.message : error,
|
|
tradeId,
|
|
subTag: orderSubTag,
|
|
timestamp: Date.now()
|
|
});
|
|
}
|
|
|
|
return { success: false, error: error.message };
|
|
} finally {
|
|
await releaseLockIfHeld();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Polls the exchange for order status until filled, cancelled, or max attempts reached.
|
|
*/
|
|
private async waitForFill(orderId: string, symbol: string, type: string): Promise<any> {
|
|
// Market orders usually fill instantly, but verify anyway
|
|
const maxAttempts = type === 'market' ? 3 : config.ORDER_POLL_MAX_ATTEMPTS;
|
|
const pollInterval = config.ORDER_POLL_INTERVAL_MS;
|
|
|
|
for (let i = 0; i < maxAttempts; i++) {
|
|
try {
|
|
const getOrder = this.exchange.getOrder;
|
|
if (getOrder) {
|
|
const order = await this.instrumentExchangeCall('get_order', () => getOrder.call(this.exchange, orderId));
|
|
if (order) {
|
|
const status = order.status?.toLowerCase();
|
|
if (status === 'filled' || status === 'partially_filled') {
|
|
logger.info(`[Executor] ✅ Order ${orderId} verified: ${status} @ $${order.filled_avg_price}`);
|
|
return order;
|
|
}
|
|
if (status === 'canceled' || status === 'expired' || status === 'rejected') {
|
|
logger.warn(`[Executor] ⌠Order ${orderId} terminal status: ${status}`);
|
|
return order;
|
|
}
|
|
logger.info(`[Executor] â³ Order ${orderId} status: ${status} (attempt ${i + 1}/${maxAttempts})`);
|
|
}
|
|
} else {
|
|
// Exchange doesn't support getOrder, assume filled
|
|
return { status: 'filled', filled_avg_price: 0 };
|
|
}
|
|
} catch (e: any) {
|
|
logger.warn(`[Executor] Poll error for ${orderId}: ${e.message}`);
|
|
}
|
|
|
|
if (i < maxAttempts - 1) {
|
|
await new Promise(resolve => setTimeout(resolve, pollInterval));
|
|
}
|
|
}
|
|
|
|
// --- TIMEOUT HANDLER (AUTO-CANCEL) ---
|
|
logger.warn(`[Executor] âš ï¸ Order ${orderId} timed out after ${maxAttempts} polls. Attempting to CANCEL...`);
|
|
|
|
try {
|
|
const cancelOrder = this.exchange.cancelOrder;
|
|
if (cancelOrder) {
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
const cancelled = await this.instrumentExchangeCall('cancel_order', () => cancelOrder.call(this.exchange, orderId, tradeSymbol));
|
|
|
|
if (cancelled) {
|
|
logger.info(`[Executor] 🛑 Order ${orderId} CANCELLED by bot due to timeout.`);
|
|
return { status: 'canceled' };
|
|
}
|
|
} else {
|
|
logger.warn(`[Executor] Connector does not support cancelOrder. Leaving order ${orderId} open.`);
|
|
}
|
|
} catch (e: any) {
|
|
logger.error(`[Executor] Failed to cancel timeout order ${orderId}: ${e.message}`);
|
|
}
|
|
|
|
logger.warn(`[Executor] âš ï¸ Order ${orderId} status unknown/uncancelled. Checking exchange position as fallback...`);
|
|
// Final fallback: check if exchange has the position
|
|
try {
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
const pos = await this.exchange.getPosition(tradeSymbol);
|
|
if (pos) {
|
|
return { status: 'filled', filled_avg_price: parseFloat(pos.avg_entry_price), filled_qty: pos.qty };
|
|
}
|
|
} catch (e) { /* ignore */ }
|
|
|
|
return { status: 'unknown' };
|
|
}
|
|
|
|
private async releaseLedgerReservation(profileId?: string, amount?: number): Promise<void> {
|
|
const ledgerProfileId = this.getLedgerProfileId(profileId);
|
|
if (!ledgerProfileId || !amount || amount <= 0) return;
|
|
await capitalLedger.releaseOrderReservation(ledgerProfileId, amount);
|
|
}
|
|
|
|
public async releasePendingOrderReservation(pending?: PendingOrder): Promise<void> {
|
|
if (!pending) return;
|
|
await this.releaseLedgerReservation(pending.profileId, pending.reservedAmount);
|
|
}
|
|
|
|
public async finalizeEntryReservation(pending: PendingOrder | undefined, fillPrice: number, filledQty: number): Promise<void> {
|
|
if (!pending) return;
|
|
const ledgerProfileId = this.getLedgerProfileId(pending.profileId);
|
|
if (!ledgerProfileId) return;
|
|
|
|
const reservedAmount = Number.isFinite(pending.reservedAmount ?? 0) ? pending.reservedAmount || 0 : 0;
|
|
const reliablePrice = fillPrice > 0 ? fillPrice : pending.requestedPrice;
|
|
if (!Number.isFinite(reliablePrice) || reliablePrice <= 0 || !Number.isFinite(filledQty) || filledQty <= 0) {
|
|
if (reservedAmount > 0) {
|
|
await capitalLedger.releaseOrderReservation(ledgerProfileId, reservedAmount);
|
|
}
|
|
return;
|
|
}
|
|
|
|
const notional = filledQty * reliablePrice;
|
|
if (reservedAmount > 0) {
|
|
await capitalLedger.releaseOrderReservation(ledgerProfileId, reservedAmount);
|
|
}
|
|
if (notional > 0) {
|
|
await capitalLedger.adjustPositionReservation(ledgerProfileId, notional);
|
|
}
|
|
const delta = Number((notional - reservedAmount).toFixed(8));
|
|
const deltaWarnThreshold = Math.max(5, reservedAmount * 0.05);
|
|
if (Math.abs(delta) >= deltaWarnThreshold) {
|
|
logger.warn(`[Executor] Entry settlement delta for ${pending.symbol}: reserved=${reservedAmount.toFixed(2)} filledNotional=${notional.toFixed(2)} delta=${delta.toFixed(2)}`);
|
|
}
|
|
}
|
|
|
|
public async reconcileEntryFill(order: any, fillPrice: number, filledQty: number): Promise<void> {
|
|
const orderId = String(order.order_id || order.id || order.orderId || '').trim();
|
|
if (!orderId) return;
|
|
|
|
const resolvedSymbol = String(order.symbol || order.symbol_name || '').trim();
|
|
const resolvedQty = Number.isFinite(filledQty) && filledQty > 0 ? filledQty : getReconciliationFillQty(order);
|
|
const resolvedPrice = Number.isFinite(fillPrice) && fillPrice > 0 ? fillPrice : getReconciliationFillPrice(order);
|
|
if (resolvedQty <= 0 || resolvedPrice <= 0 || !resolvedSymbol) return;
|
|
|
|
const pending: PendingOrder = {
|
|
orderId,
|
|
symbol: resolvedSymbol,
|
|
side: this.normalizeSignalDirection(order.side),
|
|
qty: resolvedQty,
|
|
type: this.normalizeOrderType(order.type || order.order_type),
|
|
requestedPrice: resolvedPrice,
|
|
stopLoss: Number(order.stop_loss || 0),
|
|
takeProfit: Number(order.take_profit || 0),
|
|
tradeId: order.trade_id || order.tradeId,
|
|
userId: order.user_id || this.userId,
|
|
profileId: order.profile_id || this.profileId,
|
|
subTag: extractOrderSubTag(order) || undefined,
|
|
placedAt: Number(order.timestamp || Date.now()),
|
|
action: 'ENTRY',
|
|
reservedAmount: Math.max(resolvedQty * resolvedPrice, this.computeReservedAmount(order, resolvedPrice))
|
|
};
|
|
|
|
this.pendingOrders.set(orderId, pending);
|
|
try {
|
|
await this.finalizeEntryReservation(pending, resolvedPrice, resolvedQty);
|
|
} finally {
|
|
this.pendingOrders.delete(orderId);
|
|
}
|
|
}
|
|
|
|
public async reconcileExitFill(order: any, fillPrice: number, filledQty: number): Promise<void> {
|
|
const symbol = String(order.symbol || order.symbol_name || '').trim();
|
|
if (!symbol) return;
|
|
const qty = Number.isFinite(filledQty) && filledQty > 0 ? filledQty : getReconciliationFillQty(order);
|
|
const price = Number.isFinite(fillPrice) && fillPrice > 0 ? fillPrice : getReconciliationFillPrice(order);
|
|
if (qty <= 0 || price <= 0) return;
|
|
const tradeId = String(order.trade_id || order.tradeId || '').trim();
|
|
|
|
await this.applyExitFill(symbol, price, qty, 'Reconciliation fill', tradeId || undefined, order.side);
|
|
}
|
|
|
|
public async reconcileCancel(order: any): Promise<void> {
|
|
const orderId = String(order.order_id || order.id || order.orderId || '').trim();
|
|
if (orderId && this.pendingOrders.has(orderId)) {
|
|
const pending = this.pendingOrders.get(orderId);
|
|
await this.releasePendingOrderReservation(pending);
|
|
this.pendingOrders.delete(orderId);
|
|
return;
|
|
}
|
|
|
|
const ledgerProfileId = this.getLedgerProfileId(order.profile_id || this.profileId);
|
|
if (!ledgerProfileId) return;
|
|
|
|
const reservedAmount = this.computeReservedAmount(order, getReconciliationFillPrice(order));
|
|
if (reservedAmount > 0) {
|
|
await capitalLedger.releaseOrderReservation(ledgerProfileId, reservedAmount);
|
|
logger.info('Cancel applied', {
|
|
event: 'reconciliation_cancel',
|
|
profileId: ledgerProfileId,
|
|
orderId,
|
|
symbol: String(order.symbol || order.symbol_name || ''),
|
|
reservedAmount
|
|
});
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Closes an existing position.
|
|
*/
|
|
public async closePosition(
|
|
symbol: string,
|
|
reason: string = 'Exit Signal',
|
|
tradeId?: string
|
|
): Promise<{ success: boolean, exitPrice?: number, error?: string }> {
|
|
const selected = this.resolvePositionSelection(symbol, tradeId);
|
|
if (!selected) return { success: false, error: "No active position" };
|
|
const pos = selected.position;
|
|
const positionTradeId = String(pos.tradeId || '').trim();
|
|
|
|
try {
|
|
const existingExit = this.getExitLifecycle(symbol);
|
|
if (existingExit.state === 'initiated' || existingExit.state === 'order_placed' || existingExit.state === 'verifying') {
|
|
logger.warn(`[Executor] Exit already in progress for ${symbol}. Current state: ${existingExit.state}`);
|
|
return { success: false, error: `Exit already in progress (${existingExit.state})` };
|
|
}
|
|
if (existingExit.state === 'quarantined') {
|
|
logger.warn(`[Executor] Exit blocked for ${symbol}: lifecycle is quarantined and requires manual reconciliation.`);
|
|
return { success: false, error: 'EXIT_REQUIRES_MANUAL_RECONCILIATION' };
|
|
}
|
|
|
|
this.setExitLifecycle(symbol, 'initiated', reason);
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
const exitSide = pos.side === SignalDirection.BUY ? 'sell' : 'buy';
|
|
const exitSubTag = this.buildOrderSubTag(positionTradeId || undefined, 'EXIT');
|
|
const exitClientOrderId = positionTradeId
|
|
? `bytelyst-${pos.profileId || this.profileId || 'global'}-${positionTradeId}-exit`
|
|
: undefined;
|
|
if (exitSubTag) {
|
|
logger.info('[Executor] Alpaca sub-tag prepared', {
|
|
event: 'alpaca_subtag_prepared',
|
|
profileId: pos.profileId || this.profileId,
|
|
userId: pos.userId || this.userId,
|
|
symbol,
|
|
tradeId: positionTradeId,
|
|
intent: 'EXIT',
|
|
subTag: exitSubTag
|
|
});
|
|
}
|
|
if (this.hasPendingAction(symbol, 'EXIT') || await this.hasActiveTradeId(positionTradeId)) {
|
|
logger.warn(`[Executor] Duplicate EXIT request blocked for ${symbol}`);
|
|
return { success: false, error: 'Duplicate exit request blocked (idempotency window)' };
|
|
}
|
|
|
|
logger.info(`[Executor] 🚪 Closing ${symbol} | Reason: ${reason}`);
|
|
|
|
// --- Pre-flight SELL Guard (Ghost Position Check) ---
|
|
if (exitSide === 'sell') {
|
|
const currentPos = await this.instrumentExchangeCall('get_position', () => this.exchange.getPosition(tradeSymbol));
|
|
const exchangeQty = Math.abs(Number(currentPos?.qty || 0));
|
|
|
|
if (exchangeQty < pos.size) {
|
|
if (exchangeQty > 0) {
|
|
logger.warn(`[Hardening] Partial SELL Exit for ${symbol}: adjusting order qty from ${pos.size} to available ${exchangeQty}. Remaining qty stays open pending fill evidence.`);
|
|
// We do NOT mutate pos.size here to avoid capital leaks in finalizeTrade.
|
|
// We use a local variable for the order volume.
|
|
} else {
|
|
logger.error(`[Guardrail] Aborting SELL Exit for ${symbol}: Insufficient exchange balance (Requested: ${pos.size}, Exchange: ${exchangeQty}).`);
|
|
if (config.REQUIRE_EXCHANGE_FILL_EVIDENCE_FOR_AUTO_CLOSE) {
|
|
this.setExitLifecycle(symbol, 'quarantined', 'EXCHANGE_STATE_MISMATCH', `Exchange inventory: ${exchangeQty}`);
|
|
observabilityService.emitEvent({
|
|
type: 'EXIT_FILL_COHERENCE_VIOLATION',
|
|
severity: 'ERROR',
|
|
message: `Exit blocked for ${symbol}: exchange inventory is flat while DB position remains open. Manual reconciliation required.`,
|
|
profileId: pos.profileId || this.profileId,
|
|
userId: pos.userId || this.userId,
|
|
symbol
|
|
});
|
|
if (this.apiServer) {
|
|
this.apiServer.recordOrderFailure({
|
|
profileId: pos.profileId || this.profileId,
|
|
userId: pos.userId || this.userId,
|
|
symbol,
|
|
side: 'SELL',
|
|
qty: pos.size,
|
|
reason: 'EXCHANGE_STATE_MISMATCH',
|
|
tradeId: positionTradeId,
|
|
subTag: exitSubTag,
|
|
timestamp: Date.now()
|
|
});
|
|
}
|
|
return { success: false, error: 'EXCHANGE_STATE_MISMATCH_MANUAL_REVIEW' };
|
|
}
|
|
|
|
logger.warn(`[Guardrail] REQUIRE_EXCHANGE_FILL_EVIDENCE_FOR_AUTO_CLOSE=false; applying legacy local finalize fallback for ${symbol}.`);
|
|
await this.finalizeTrade(symbol, pos.entryPrice, 'EXCHANGE_STATE_MISMATCH', positionTradeId);
|
|
this.setExitLifecycle(symbol, 'failed', 'EXCHANGE_STATE_MISMATCH', `Exchange inventory: ${exchangeQty}`);
|
|
return { success: false, error: 'EXCHANGE_STATE_MISMATCH' };
|
|
}
|
|
}
|
|
}
|
|
|
|
// Using helper to get available qty for closure
|
|
const exchangeQty = await this.getExchangeAvailableQty(tradeSymbol);
|
|
const effectiveOrderQty = (exitSide === 'sell')
|
|
? Math.min(pos.size, exchangeQty)
|
|
: pos.size;
|
|
|
|
// Fetch current price for logging if needed, or rely on execution
|
|
// We usually just execute market exit
|
|
const order = await this.instrumentExchangeCall('place_order', () => this.exchange.placeOrder(
|
|
tradeSymbol,
|
|
exitSide,
|
|
effectiveOrderQty,
|
|
'market',
|
|
undefined,
|
|
undefined,
|
|
undefined,
|
|
exitClientOrderId,
|
|
{
|
|
subTag: exitSubTag,
|
|
profileId: pos.profileId || this.profileId,
|
|
tradeId: positionTradeId || undefined,
|
|
intent: 'EXIT'
|
|
}
|
|
));
|
|
|
|
if (!order) {
|
|
return { success: false, error: "Order failed" };
|
|
}
|
|
if (exitClientOrderId) {
|
|
order.client_order_id = order.client_order_id || exitClientOrderId;
|
|
}
|
|
if (exitSubTag) {
|
|
order.subtag = order.subtag || exitSubTag;
|
|
order.sub_tag = order.sub_tag || exitSubTag;
|
|
}
|
|
this.setExitLifecycle(symbol, 'order_placed', reason, undefined, order.id);
|
|
|
|
// --- Order Fill Verification ---
|
|
this.setExitLifecycle(symbol, 'verifying', reason, undefined, order.id);
|
|
const verifiedOrder = await this.waitForFill(order.id, symbol, 'market');
|
|
const verifiedStatus = (verifiedOrder?.status || '').toLowerCase();
|
|
const exitPrice = Number(verifiedOrder?.filled_avg_price || order.filled_avg_price || 0);
|
|
const rawFilledQty = this.getFilledQuantity(verifiedOrder);
|
|
const exchangeFilledQty = Number.isFinite(rawFilledQty as number) && Number(rawFilledQty) > 0
|
|
? Number(rawFilledQty)
|
|
: undefined;
|
|
const normalizedFilledQty = exchangeFilledQty
|
|
? Math.min(exchangeFilledQty, pos.size)
|
|
: undefined;
|
|
const persistedExitQty = exchangeFilledQty && exchangeFilledQty > 0
|
|
? exchangeFilledQty
|
|
: pos.size;
|
|
const finalUserId = pos.userId || this.userId;
|
|
|
|
// Log the Exit with same trade_id for full cycle tracing
|
|
await this.logOrderToDb(order, symbol, exitSide, persistedExitQty, exitPrice, 'market', 0, 0, finalUserId, positionTradeId, 'EXIT');
|
|
this.updateDashboardOrder(order, symbol, exitSide, persistedExitQty, exitPrice, 'market', positionTradeId, 'EXIT');
|
|
|
|
// --- Track for background sync ---
|
|
this.pendingOrders.set(order.id, {
|
|
orderId: order.id,
|
|
symbol,
|
|
side: pos.side === SignalDirection.BUY ? SignalDirection.SELL : SignalDirection.BUY, // opposite for exit
|
|
qty: persistedExitQty,
|
|
type: 'market',
|
|
requestedPrice: exitPrice,
|
|
stopLoss: 0,
|
|
takeProfit: 0,
|
|
tradeId: positionTradeId,
|
|
userId: finalUserId,
|
|
subTag: exitSubTag,
|
|
placedAt: Date.now(),
|
|
action: 'EXIT'
|
|
});
|
|
|
|
// Remove from tracking as this close path handles terminal state immediately
|
|
this.pendingOrders.delete(order.id);
|
|
|
|
// Do NOT finalize trade locally unless exchange confirms a fill.
|
|
if (!verifiedOrder || ['canceled', 'expired', 'rejected', 'unknown'].includes(verifiedStatus)) {
|
|
logger.warn(`[Executor] âš ï¸ Exit order ${order.id} for ${symbol} ended as ${verifiedStatus || 'unknown'}. Keeping local position open.`);
|
|
await runtimeOrderRepository.updateOrderStatus(order.id, verifiedStatus || 'unknown');
|
|
this.setExitLifecycle(
|
|
symbol,
|
|
verifiedStatus === 'unknown' ? 'quarantined' : 'failed',
|
|
reason,
|
|
`verified_status=${verifiedStatus || 'unknown'}`,
|
|
order.id
|
|
);
|
|
await this.notifier.sendAlert(`âš ï¸ **EXIT NOT CONFIRMED**\nSymbol: ${symbol}\nStatus: ${verifiedStatus || 'unknown'}\nAction: Manual review required`);
|
|
return { success: false, error: `Exit order ${verifiedStatus || 'unknown'}` };
|
|
}
|
|
|
|
// ✅ Update order status in DB for confirmed exit fills
|
|
if (verifiedStatus === 'partially_filled' && (!normalizedFilledQty || normalizedFilledQty <= 0)) {
|
|
await runtimeOrderRepository.updateOrderStatus(order.id, verifiedStatus, new Date(), exitPrice > 0 ? exitPrice : undefined);
|
|
this.setExitLifecycle(symbol, 'quarantined', reason, 'partial_fill_missing_qty', order.id);
|
|
await this.notifier.sendAlert(`PARTIAL EXIT QUARANTINED\nSymbol: ${symbol}\nOrder: ${order.id}\nAction: Fill qty missing, manual review required`);
|
|
return { success: false, error: 'Partial exit fill qty missing' };
|
|
}
|
|
|
|
await runtimeOrderRepository.updateOrderStatus(
|
|
order.id,
|
|
verifiedStatus || 'filled',
|
|
new Date(),
|
|
exitPrice > 0 ? exitPrice : undefined,
|
|
persistedExitQty
|
|
);
|
|
const applied = await this.applyExitFill(
|
|
symbol,
|
|
exitPrice,
|
|
normalizedFilledQty || (verifiedStatus === 'filled' ? pos.size : undefined),
|
|
reason,
|
|
positionTradeId,
|
|
order.side
|
|
);
|
|
|
|
if (!applied.success) {
|
|
this.setExitLifecycle(symbol, 'failed', reason, applied.error || 'exit_fill_apply_failed', order.id);
|
|
return { success: false, error: applied.error || 'Failed to apply exit fill' };
|
|
}
|
|
|
|
if (!applied.fullyClosed) {
|
|
this.setExitLifecycle(symbol, 'idle', reason, `partial_exit_remaining=${applied.remainingSize}`, order.id);
|
|
await this.notifier.sendAlert(`PARTIAL EXIT FILLED\nSymbol: ${symbol}\nFilled Qty: ${applied.appliedQty}\nRemaining Qty: ${applied.remainingSize}\nPrice: ${exitPrice}`);
|
|
logger.info('Exit partial fill', {
|
|
event: 'exit_partial_fill',
|
|
symbol,
|
|
tradeId: positionTradeId,
|
|
profileId: this.profileId,
|
|
filledQty: applied.appliedQty,
|
|
remainingQty: applied.remainingSize,
|
|
price: exitPrice
|
|
});
|
|
return { success: true, exitPrice };
|
|
}
|
|
|
|
this.setExitLifecycle(symbol, 'filled', reason, `verified_status=${verifiedStatus || 'filled'}`, order.id);
|
|
logger.info('EXIT filled', {
|
|
event: 'exit_filled',
|
|
symbol,
|
|
tradeId: positionTradeId,
|
|
profileId: this.profileId,
|
|
filledQty: persistedExitQty,
|
|
price: exitPrice
|
|
});
|
|
await this.notifier.sendAlert(`POSITION CLOSED\nSymbol: ${symbol}\nReason: ${reason}\nPrice: ${exitPrice}`);
|
|
return { success: true, exitPrice };
|
|
} catch (error: any) {
|
|
logger.error(`[Executor] ⌠Close Failed: ${error.message}`);
|
|
this.setExitLifecycle(symbol, 'failed', reason, error.message);
|
|
await this.notifier.sendAlert(`⌠**CLOSE FAILED**\nSymbol: ${symbol}\nError: ${error.message}`);
|
|
return { success: false, error: error.message };
|
|
}
|
|
}
|
|
|
|
public async applyExitFill(
|
|
symbol: string,
|
|
exitPrice: number,
|
|
fillQty: number | undefined,
|
|
reason: string,
|
|
tradeId?: string,
|
|
fillSide?: string
|
|
): Promise<ExitFillApplyResult> {
|
|
const selected = this.resolvePositionSelection(symbol, tradeId);
|
|
if (!selected) {
|
|
return {
|
|
success: false,
|
|
fullyClosed: false,
|
|
appliedQty: 0,
|
|
remainingSize: 0,
|
|
error: 'No active position'
|
|
};
|
|
}
|
|
const pos = selected.position;
|
|
const selectedTradeId = String(pos.tradeId || '').trim();
|
|
|
|
// FIX-05: Directional coherence guard
|
|
const exitSideExpected = pos.side === SignalDirection.BUY ? 'SELL' : 'BUY';
|
|
const fillSideNormalized = String(fillSide || '').trim().toUpperCase();
|
|
if (fillSideNormalized && fillSideNormalized !== exitSideExpected && fillSideNormalized !== 'UNKNOWN') {
|
|
logger.error(`[Executor] ❌ applyExitFill coherence violation for ${symbol}: position is ${pos.side}, but fill side is ${fillSideNormalized}. Aborting exit application.`);
|
|
observabilityService.emitEvent({
|
|
type: 'EXIT_FILL_COHERENCE_VIOLATION',
|
|
severity: 'ERROR',
|
|
message: `Exit fill side ${fillSideNormalized} does not match expected ${exitSideExpected} for ${pos.side} position.`,
|
|
profileId: this.profileId,
|
|
symbol
|
|
});
|
|
return { success: false, fullyClosed: false, appliedQty: 0, remainingSize: pos.size, error: 'coherence_violation' };
|
|
}
|
|
|
|
const requestedFillQty = Number(fillQty);
|
|
if (!Number.isFinite(requestedFillQty) || requestedFillQty <= 0) {
|
|
return {
|
|
success: false,
|
|
fullyClosed: false,
|
|
appliedQty: 0,
|
|
remainingSize: pos.size,
|
|
error: 'Missing exit fill qty'
|
|
};
|
|
}
|
|
|
|
const appliedQty = Math.min(requestedFillQty, pos.size);
|
|
const remainingSize = Math.max(0, Number((pos.size - appliedQty).toFixed(8)));
|
|
const resolvedExitPrice = Number.isFinite(exitPrice) && exitPrice > 0 ? exitPrice : pos.entryPrice;
|
|
|
|
// Fully closed lifecycle: finalize and clear local position state.
|
|
if (remainingSize <= 1e-8) {
|
|
await this.finalizeTrade(symbol, resolvedExitPrice, reason, selectedTradeId);
|
|
return {
|
|
success: true,
|
|
fullyClosed: true,
|
|
appliedQty,
|
|
remainingSize: 0
|
|
};
|
|
}
|
|
|
|
// Partial exit lifecycle: persist realized slice and keep monitoring remainder.
|
|
const pnl = (resolvedExitPrice - pos.entryPrice) * appliedQty * (pos.side === SignalDirection.BUY ? 1 : -1);
|
|
const pnlPercent = pos.entryPrice > 0
|
|
? ((resolvedExitPrice - pos.entryPrice) / pos.entryPrice) * 100 * (pos.side === SignalDirection.BUY ? 1 : -1)
|
|
: 0;
|
|
const finalUserId = pos.userId || this.userId;
|
|
const normalizedTradeId = selectedTradeId;
|
|
let canLogPartial = true;
|
|
|
|
if (finalUserId !== 'global') {
|
|
if (normalizedTradeId) {
|
|
canLogPartial = await runtimeOrderRepository.hasLifecycleEntryOrder(
|
|
normalizedTradeId,
|
|
pos.profileId || this.profileId,
|
|
symbol
|
|
);
|
|
if (!canLogPartial) {
|
|
logger.warn(`[Executor] Skipping partial EXIT history log for ${symbol}: lifecycle ${normalizedTradeId} has no ENTRY chain.`);
|
|
}
|
|
}
|
|
|
|
if (canLogPartial) {
|
|
await runtimeOrderRepository.logTransaction({
|
|
user_id: finalUserId,
|
|
profile_id: pos.profileId || this.profileId,
|
|
symbol,
|
|
side: pos.side,
|
|
entry_price: pos.entryPrice,
|
|
exit_price: resolvedExitPrice,
|
|
size: appliedQty,
|
|
pnl,
|
|
pnl_percent: pnlPercent,
|
|
reason: `${reason} (Partial Exit)`,
|
|
timestamp: Date.now(),
|
|
stop_loss: pos.stopLoss || undefined,
|
|
take_profit: pos.takeProfit || undefined,
|
|
trade_id: pos.tradeId,
|
|
source: 'BOT'
|
|
});
|
|
}
|
|
}
|
|
|
|
const ledgerProfileId = this.getLedgerProfileId(pos.profileId || this.profileId);
|
|
// Only mutate capital ledger when this partial exit has a valid lifecycle ENTRY chain.
|
|
if (ledgerProfileId && canLogPartial) {
|
|
const reduction = appliedQty * (pos.entryPrice || 0);
|
|
if (reduction > 0) {
|
|
await capitalLedger.adjustPositionReservation(ledgerProfileId, -reduction);
|
|
}
|
|
if (Number.isFinite(pnl)) {
|
|
await capitalLedger.recordRealizedPnl(ledgerProfileId, pnl);
|
|
}
|
|
} else if (ledgerProfileId && !canLogPartial) {
|
|
logger.warn(`[Executor] Skipping partial EXIT ledger mutation for ${symbol}: lifecycle ${normalizedTradeId || 'unknown'} has no ENTRY chain.`);
|
|
}
|
|
|
|
this.upsertPosition(symbol, {
|
|
...pos,
|
|
symbol,
|
|
size: remainingSize,
|
|
peakPrice: resolvedExitPrice
|
|
});
|
|
|
|
if (this.apiServer) {
|
|
const allPos = this.getAllActivePositions();
|
|
this.apiServer.updatePositions(allPos, this.profileId || 'global');
|
|
}
|
|
|
|
return {
|
|
success: true,
|
|
fullyClosed: false,
|
|
appliedQty,
|
|
remainingSize
|
|
};
|
|
}
|
|
|
|
// --- Aliases for Compatibility/Clarity ---
|
|
public async executeExit(symbol: string, currentPrice: number, reason: string, tradeId?: string) {
|
|
// We ignore currentPrice for market order, but could log it
|
|
return this.closePosition(symbol, reason, tradeId);
|
|
}
|
|
|
|
public async markTradeComplete(symbol: string, exitPrice: number, reason: string = 'Target Reached', tradeId?: string) {
|
|
return await this.finalizeTrade(symbol, exitPrice, reason, tradeId);
|
|
}
|
|
|
|
/**
|
|
* Consolidates trade history logic.
|
|
* Clears local state and starts cooldown.
|
|
*/
|
|
public async finalizeTrade(symbol: string, exitPrice: number, reason: string, tradeId?: string) {
|
|
const selected = this.resolvePositionSelection(symbol, tradeId);
|
|
let pos = selected?.position;
|
|
const normalizedInputTradeId = String(tradeId || '').trim();
|
|
|
|
// --- RECOVERY: If lost from memory (restart gap), try to recover ENTRY from DB ---
|
|
if (!pos) {
|
|
logger.info(`[Executor] 🔠Position for ${symbol} not in memory. Attempting DB recovery for history log...`);
|
|
try {
|
|
// Look for the latest ENTRY order for this symbol/profile
|
|
if (this.profileId) {
|
|
const lastEntry = await runtimeOrderRepository.getLatestEntryOrder(this.profileId, symbol, this.userId);
|
|
|
|
if (lastEntry) {
|
|
pos = {
|
|
symbol,
|
|
side: lastEntry.side as SignalDirection,
|
|
entryPrice: lastEntry.price,
|
|
size: lastEntry.qty,
|
|
stopLoss: lastEntry.stop_loss,
|
|
takeProfit: lastEntry.take_profit,
|
|
peakPrice: lastEntry.price,
|
|
userId: lastEntry.user_id,
|
|
profileId: lastEntry.profile_id,
|
|
tradeId: lastEntry.trade_id
|
|
};
|
|
logger.info(`[Executor] 🧠Recovered entry details for ${symbol} (Price: ${pos.entryPrice})`);
|
|
}
|
|
}
|
|
} catch (e) {
|
|
logger.warn(`[Executor] Failed to recover entry details for ${symbol}: ${e}`);
|
|
}
|
|
}
|
|
|
|
if (pos && exitPrice) {
|
|
const profileScope = pos.profileId || this.profileId;
|
|
const normalizedTradeId = String(pos.tradeId || normalizedInputTradeId).trim();
|
|
|
|
let hasEntryChain = true;
|
|
if (normalizedTradeId) {
|
|
hasEntryChain = await runtimeOrderRepository.hasLifecycleEntryOrder(normalizedTradeId, profileScope, symbol);
|
|
if (!hasEntryChain) {
|
|
logger.warn(`[Executor] Suppressing finalize history for ${symbol}: lifecycle ${normalizedTradeId} has no ENTRY chain.`);
|
|
}
|
|
}
|
|
|
|
let alreadyFinalized = false;
|
|
if (normalizedTradeId && hasEntryChain) {
|
|
alreadyFinalized = await runtimeOrderRepository.hasFinalizedTradeHistory(normalizedTradeId, profileScope, symbol);
|
|
}
|
|
|
|
const pnl = (exitPrice - pos.entryPrice) * pos.size * (pos.side === SignalDirection.BUY ? 1 : -1);
|
|
const pnlPercent = ((exitPrice - pos.entryPrice) / pos.entryPrice) * 100 * (pos.side === SignalDirection.BUY ? 1 : -1);
|
|
const ledgerProfileId = this.getLedgerProfileId(profileScope);
|
|
const canMutateLedger = hasEntryChain && !alreadyFinalized;
|
|
if (
|
|
canMutateLedger
|
|
&& ledgerProfileId
|
|
&& Number.isFinite(pos.size)
|
|
&& pos.size > 0
|
|
&& Number.isFinite(pos.entryPrice)
|
|
&& pos.entryPrice > 0
|
|
) {
|
|
const releaseNotional = pos.size * pos.entryPrice;
|
|
await capitalLedger.adjustPositionReservation(ledgerProfileId, -releaseNotional);
|
|
await capitalLedger.recordRealizedPnl(ledgerProfileId, pnl);
|
|
} else if (ledgerProfileId && !canMutateLedger) {
|
|
logger.warn(`[Executor] Skipping finalize ledger mutation for ${symbol} (trade=${normalizedTradeId || 'unknown'}): hasEntryChain=${hasEntryChain}, alreadyFinalized=${alreadyFinalized}.`);
|
|
}
|
|
|
|
if (alreadyFinalized) {
|
|
logger.warn(`[Executor] Duplicate finalize suppressed for ${symbol} (trade=${normalizedTradeId}).`);
|
|
} else if (hasEntryChain) {
|
|
if (this.apiServer) {
|
|
this.apiServer.addHistory({
|
|
symbol,
|
|
side: pos.side,
|
|
entryPrice: pos.entryPrice,
|
|
exitPrice,
|
|
size: pos.size,
|
|
pnl,
|
|
pnlPercent,
|
|
reason,
|
|
timestamp: Date.now(),
|
|
profileId: profileScope,
|
|
source: 'BOT',
|
|
trade_id: pos.tradeId
|
|
});
|
|
}
|
|
|
|
const finalUserId = pos.userId || this.userId;
|
|
if (finalUserId !== 'global') {
|
|
await runtimeOrderRepository.logTransaction({
|
|
user_id: finalUserId,
|
|
profile_id: profileScope,
|
|
symbol,
|
|
side: pos.side,
|
|
entry_price: pos.entryPrice,
|
|
exit_price: exitPrice,
|
|
size: pos.size,
|
|
pnl,
|
|
pnl_percent: pnlPercent,
|
|
reason,
|
|
timestamp: Date.now(),
|
|
stop_loss: pos.stopLoss || undefined,
|
|
take_profit: pos.takeProfit || undefined,
|
|
trade_id: pos.tradeId,
|
|
source: 'BOT'
|
|
});
|
|
}
|
|
}
|
|
} else {
|
|
logger.warn(`[Executor] Could not finalize trade for ${symbol}: Missing position data.`);
|
|
}
|
|
|
|
this.removePosition(symbol, normalizedInputTradeId || pos?.tradeId);
|
|
const hasRemainingSymbolPositions = this.getPositionsForSymbol(symbol).length > 0;
|
|
if (!hasRemainingSymbolPositions) {
|
|
this.exitLifecycle.delete(symbol);
|
|
this.cooldowns.set(symbol, Date.now());
|
|
}
|
|
if (this.apiServer) {
|
|
const allPos = this.getAllActivePositions();
|
|
this.apiServer.updatePositions(allPos, this.profileId || 'global');
|
|
}
|
|
logger.info(`[Executor] ✅ Trade finalized for ${symbol}. Cooldown started.`);
|
|
}
|
|
|
|
// --- Helpers ---
|
|
|
|
private async getExchangeAvailableQty(tradeSymbol: string): Promise<number> {
|
|
try {
|
|
const currentPos = await this.instrumentExchangeCall('get_position', () => this.exchange.getPosition(tradeSymbol));
|
|
return Math.abs(Number(currentPos?.qty || 0));
|
|
} catch (e) {
|
|
logger.warn(`[Executor] Failed to fetch exchange qty for ${tradeSymbol}: ${e}`);
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
private async logOrderToDb(order: any, symbol: string, side: string, qty: number, price: number, type: string, stopLoss?: number, takeProfit?: number, specificUserId?: string, tradeId?: string, action?: string) {
|
|
const finalUserId = specificUserId || this.userId;
|
|
const orderSubTag = extractOrderSubTag(order) || undefined;
|
|
if (finalUserId !== 'global') {
|
|
await runtimeOrderRepository.logOrder({
|
|
user_id: finalUserId,
|
|
profile_id: this.profileId,
|
|
order_id: order.id,
|
|
symbol,
|
|
type: normalizeOrderType(type),
|
|
side: normalizeTradeSide(side),
|
|
qty,
|
|
price: price,
|
|
status: normalizeOrderStatus(order.status || 'filled'),
|
|
timestamp: Date.now(),
|
|
stop_loss: stopLoss,
|
|
take_profit: takeProfit,
|
|
trade_id: tradeId,
|
|
action: normalizeOrderAction(action), // 'ENTRY' or 'EXIT'
|
|
sub_tag: orderSubTag
|
|
});
|
|
}
|
|
}
|
|
|
|
private updateDashboardOrder(order: any, symbol: string, side: string, qty: number, price: number, type: string, tradeId?: string, action?: string) {
|
|
// We now use broadcastOrders to send the FULL current state to avoid overwriting issues
|
|
this.broadcastOrders();
|
|
}
|
|
|
|
public broadcastOrders() {
|
|
if (!this.apiServer) return;
|
|
|
|
const orders = Array.from(this.pendingOrders.values()).map(p => ({
|
|
id: p.orderId,
|
|
symbol: p.symbol,
|
|
type: p.type === 'market' ? 'Market' : 'Limit',
|
|
side: p.side as string,
|
|
qty: p.qty,
|
|
price: p.requestedPrice,
|
|
status: 'pending_new',
|
|
timestamp: p.placedAt,
|
|
profileId: this.profileId,
|
|
source: 'BOT' as const,
|
|
trade_id: p.tradeId,
|
|
subTag: p.subTag,
|
|
action: p.action
|
|
}));
|
|
|
|
this.apiServer.updateOrders(orders, this.profileId || 'global');
|
|
}
|
|
|
|
public async syncPositions(symbols: string[]) {
|
|
logger.info('[Executor] Syncing exchange positions...');
|
|
const isDedicatedProfileScope = !!this.profileId
|
|
&& this.profileId !== 'global'
|
|
&& !this.profileId.startsWith('default-');
|
|
|
|
for (const symbol of symbols) {
|
|
try {
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
const position = await this.exchange.getPosition(tradeSymbol);
|
|
const hasExchangePosition = !!position && Math.abs(Number(position.qty || 0)) > 0;
|
|
|
|
if (isDedicatedProfileScope && this.profileId) {
|
|
const symbolCandidates = Array.from(new Set([
|
|
symbol,
|
|
tradeSymbol,
|
|
SymbolMapper.toDataSymbol(symbol, config.EXECUTION_PROVIDER),
|
|
SymbolMapper.toDataSymbol(tradeSymbol, config.EXECUTION_PROVIDER)
|
|
].filter(Boolean)));
|
|
|
|
let virtualPosition = null as Awaited<ReturnType<typeof runtimeOrderRepository.getVirtualOpenPosition>>;
|
|
for (const candidateSymbol of symbolCandidates) {
|
|
virtualPosition = await runtimeOrderRepository.getVirtualOpenPosition(this.profileId, candidateSymbol);
|
|
if (virtualPosition) break;
|
|
}
|
|
|
|
const previousSymbolPositions = this.getActivePositions(symbol);
|
|
if (!virtualPosition) {
|
|
if (!hasExchangePosition) {
|
|
if (previousSymbolPositions.length > 0) {
|
|
this.removePosition(symbol);
|
|
logger.info(`[Executor] Cleared ${previousSymbolPositions.length} local profile position(s) for ${symbol} under ${this.profileId}; exchange is flat and no virtual lifecycle remained.`);
|
|
}
|
|
} else if (previousSymbolPositions.length > 0) {
|
|
logger.warn(`[Executor] Retaining ${previousSymbolPositions.length} local profile position(s) for ${symbol} under ${this.profileId}; exchange is open but virtual lifecycle lookup returned empty (checked ${symbolCandidates.join(', ')}).`);
|
|
} else {
|
|
logger.warn(`[Executor] Skipping sync claim for ${symbol} under profile ${this.profileId}: no profile-scoped virtual open position found (checked ${symbolCandidates.join(', ')}).`);
|
|
}
|
|
continue;
|
|
}
|
|
|
|
const virtualSide = virtualPosition.side === 'BUY' ? SignalDirection.BUY : SignalDirection.SELL;
|
|
|
|
const exchangeSideRaw = (position?.side || '').toLowerCase();
|
|
const exchangeSide = (exchangeSideRaw === 'long' || exchangeSideRaw === 'buy')
|
|
? SignalDirection.BUY
|
|
: SignalDirection.SELL;
|
|
const exchangeQty = Math.abs(Number(position?.qty || 0));
|
|
|
|
if (!hasExchangePosition) {
|
|
logger.warn(`[Executor] Virtual position exists for ${symbol} under profile ${this.profileId}, but exchange is flat. Keeping virtual profile state.`);
|
|
} else {
|
|
if (exchangeSide !== virtualSide) {
|
|
logger.warn(`[Executor] Side mismatch for ${symbol} under profile ${this.profileId}: virtual=${virtualSide}, exchange=${exchangeSide}.`);
|
|
}
|
|
if (exchangeQty > 0 && virtualPosition.qty > exchangeQty + 1e-8) {
|
|
logger.warn(`[Executor] Qty mismatch for ${symbol} under profile ${this.profileId}: virtual=${virtualPosition.qty}, exchange=${exchangeQty}.`);
|
|
}
|
|
}
|
|
|
|
const virtualTradeIds = Array.from(new Set((virtualPosition.tradeIds || []).map((id) => String(id || '').trim()).filter(Boolean)));
|
|
const recoveredSlices: PositionState[] = [];
|
|
|
|
for (const tradeId of virtualTradeIds) {
|
|
const slice = await runtimeOrderRepository.getVirtualOpenPositionForTrade(
|
|
this.profileId,
|
|
virtualPosition.symbol || symbol,
|
|
tradeId
|
|
);
|
|
if (!slice) continue;
|
|
|
|
let recoveredStopLoss = Number(slice.stopLoss || 0);
|
|
let recoveredTakeProfit = Number(slice.takeProfit || 0);
|
|
if (recoveredStopLoss <= 0 || recoveredTakeProfit <= 0) {
|
|
const riskFallback = await runtimeOrderRepository.getLatestEntryRiskOrder(
|
|
this.profileId,
|
|
slice.symbol || symbol,
|
|
slice.side
|
|
);
|
|
if (riskFallback) {
|
|
const sl = Number(riskFallback.stop_loss);
|
|
const tp = Number(riskFallback.take_profit);
|
|
if (recoveredStopLoss <= 0 && Number.isFinite(sl) && sl > 0) {
|
|
recoveredStopLoss = sl;
|
|
}
|
|
if (recoveredTakeProfit <= 0 && Number.isFinite(tp) && tp > 0) {
|
|
recoveredTakeProfit = tp;
|
|
}
|
|
}
|
|
}
|
|
|
|
const existingLocal = previousSymbolPositions.find((candidate) => String(candidate.tradeId || '').trim() === tradeId);
|
|
if (existingLocal && existingLocal.side === (slice.side === 'BUY' ? SignalDirection.BUY : SignalDirection.SELL)) {
|
|
if (recoveredStopLoss <= 0 && Number(existingLocal.stopLoss) > 0) {
|
|
recoveredStopLoss = Number(existingLocal.stopLoss);
|
|
}
|
|
if (recoveredTakeProfit <= 0 && Number(existingLocal.takeProfit) > 0) {
|
|
recoveredTakeProfit = Number(existingLocal.takeProfit);
|
|
}
|
|
}
|
|
|
|
recoveredSlices.push({
|
|
symbol,
|
|
side: slice.side === 'BUY' ? SignalDirection.BUY : SignalDirection.SELL,
|
|
entryPrice: slice.entryPrice,
|
|
size: slice.qty,
|
|
stopLoss: recoveredStopLoss,
|
|
takeProfit: recoveredTakeProfit,
|
|
peakPrice: slice.entryPrice,
|
|
userId: slice.userId || this.userId,
|
|
profileId: this.profileId,
|
|
tradeId: slice.tradeId
|
|
});
|
|
|
|
// FIX-01: Alert when stop-loss cannot be recovered — position will run unprotected.
|
|
if (recoveredStopLoss <= 0) {
|
|
logger.error(`[Executor] ⚠️ RECOVERY_SL_MISSING: stopLoss=0 for ${symbol} tradeId=${slice.tradeId} profile=${this.profileId}. Position runs WITHOUT stop-loss until data is available.`);
|
|
observabilityService.emitEvent({
|
|
type: 'RECOVERY_SL_MISSING',
|
|
severity: 'ERROR',
|
|
message: `Recovered position for ${symbol} (trade=${slice.tradeId}) has stopLoss=0. Stop-loss protection is DISABLED for this trade until restart or manual correction.`,
|
|
profileId: this.profileId,
|
|
symbol
|
|
});
|
|
}
|
|
}
|
|
|
|
if (recoveredSlices.length === 0) {
|
|
const virtualSide = virtualPosition.side === 'BUY' ? SignalDirection.BUY : SignalDirection.SELL;
|
|
let recoveredStopLoss = Number(virtualPosition.stopLoss || 0);
|
|
let recoveredTakeProfit = Number(virtualPosition.takeProfit || 0);
|
|
if (recoveredStopLoss <= 0 || recoveredTakeProfit <= 0) {
|
|
const riskFallback = await runtimeOrderRepository.getLatestEntryRiskOrder(
|
|
this.profileId,
|
|
virtualPosition.symbol || symbol,
|
|
virtualPosition.side
|
|
);
|
|
if (riskFallback) {
|
|
const sl = Number(riskFallback.stop_loss);
|
|
const tp = Number(riskFallback.take_profit);
|
|
if (recoveredStopLoss <= 0 && Number.isFinite(sl) && sl > 0) {
|
|
recoveredStopLoss = sl;
|
|
}
|
|
if (recoveredTakeProfit <= 0 && Number.isFinite(tp) && tp > 0) {
|
|
recoveredTakeProfit = tp;
|
|
}
|
|
}
|
|
}
|
|
|
|
recoveredSlices.push({
|
|
symbol,
|
|
side: virtualSide,
|
|
entryPrice: virtualPosition.entryPrice,
|
|
size: virtualPosition.qty,
|
|
stopLoss: recoveredStopLoss,
|
|
takeProfit: recoveredTakeProfit,
|
|
peakPrice: virtualPosition.entryPrice,
|
|
userId: virtualPosition.userId || this.userId,
|
|
profileId: this.profileId,
|
|
tradeId: virtualPosition.tradeId
|
|
});
|
|
|
|
// FIX-01: Alert when stop-loss cannot be recovered — position will run unprotected.
|
|
if (recoveredStopLoss <= 0) {
|
|
logger.error(`[Executor] ⚠️ RECOVERY_SL_MISSING: stopLoss=0 for ${symbol} tradeId=${virtualPosition.tradeId} profile=${this.profileId}. Position runs WITHOUT stop-loss until data is available.`);
|
|
observabilityService.emitEvent({
|
|
type: 'RECOVERY_SL_MISSING',
|
|
severity: 'ERROR',
|
|
message: `Recovered position for ${symbol} (trade=${virtualPosition.tradeId}) has stopLoss=0. Stop-loss protection is DISABLED for this trade until restart or manual correction.`,
|
|
profileId: this.profileId,
|
|
symbol
|
|
});
|
|
}
|
|
}
|
|
|
|
this.removePosition(symbol);
|
|
for (const recovered of recoveredSlices) {
|
|
this.upsertPosition(symbol, recovered);
|
|
}
|
|
|
|
logger.info(`[Executor] Recovered ${recoveredSlices.length} virtual profile position(s) for ${symbol} under ${this.profileId}.`);
|
|
continue;
|
|
}
|
|
|
|
if (!hasExchangePosition) {
|
|
const hadLocalPositions = this.getPositionsForSymbol(symbol).length > 0;
|
|
if (hadLocalPositions) {
|
|
this.removePosition(symbol);
|
|
logger.info(`[Executor] Exchange has no open position for ${symbol}; local state cleared.`);
|
|
}
|
|
continue;
|
|
}
|
|
|
|
const side = (position?.side || '').toLowerCase();
|
|
const finalSide = (side === 'long' || side === 'buy') ? SignalDirection.BUY : SignalDirection.SELL;
|
|
const exchangeEntryPrice = Number(position?.avg_entry_price || 0);
|
|
const exchangeSize = Math.abs(Number(position?.qty || 0));
|
|
|
|
logger.info(`[Executor] Found existing ${side} position for ${symbol}.`);
|
|
let recoveredSl = 0;
|
|
let recoveredTp = 0;
|
|
let recoveredUserId = this.userId;
|
|
let recoveredTradeId = this.buildDeterministicSyncTradeId(symbol); // Default if not found
|
|
let recoveredEntryPrice = exchangeEntryPrice;
|
|
let recoveredSize = exchangeSize;
|
|
|
|
// Try to recover from DB
|
|
try {
|
|
// Priority 1: Find the FILLED ENTRY for this specific symbol/user/profile
|
|
// This ensures we link to the start of the trade lifecycle
|
|
const entryOrder = await runtimeOrderRepository.getLatestFilledEntry(this.userId, symbol, this.profileId);
|
|
|
|
if (entryOrder) {
|
|
if (entryOrder.trade_id) {
|
|
recoveredTradeId = entryOrder.trade_id;
|
|
logger.info(`[Executor] Recovered Trade ID ${recoveredTradeId} from ENTRY order for ${symbol}`);
|
|
}
|
|
recoveredUserId = entryOrder.user_id;
|
|
recoveredSl = entryOrder.stop_loss || 0;
|
|
recoveredTp = entryOrder.take_profit || 0;
|
|
} else {
|
|
const scopedEntry = this.profileId
|
|
? await runtimeOrderRepository.getLatestEntryOrder(this.profileId, symbol, this.userId)
|
|
: null;
|
|
|
|
if (scopedEntry) {
|
|
recoveredSl = scopedEntry.stop_loss || 0;
|
|
recoveredTp = scopedEntry.take_profit || 0;
|
|
recoveredUserId = scopedEntry.user_id || recoveredUserId;
|
|
if (scopedEntry.trade_id) {
|
|
recoveredTradeId = scopedEntry.trade_id;
|
|
logger.info(`[Executor] Recovered Trade ID ${recoveredTradeId} for ${symbol} from profile-scoped entry fallback`);
|
|
}
|
|
} else {
|
|
// Legacy/global fallback: Check the very last order (might be a modification or partial fill)
|
|
const lastOrder = await runtimeOrderRepository.getLatestOrder(this.userId, symbol);
|
|
if (lastOrder) {
|
|
recoveredSl = lastOrder.stop_loss || 0;
|
|
recoveredTp = lastOrder.take_profit || 0;
|
|
recoveredUserId = lastOrder.user_id;
|
|
|
|
if (lastOrder.trade_id) {
|
|
recoveredTradeId = lastOrder.trade_id;
|
|
logger.info(`[Executor] Recovered Trade ID ${recoveredTradeId} for ${symbol} from latest order (fallback)`);
|
|
}
|
|
logger.info(`[Executor] Recovered SL/TP for ${symbol} from DB: SL=${recoveredSl}, TP=${recoveredTp}`);
|
|
} else {
|
|
logger.warn(`[Executor] No history found for ${symbol}. Using synthetic Trade ID: ${recoveredTradeId}`);
|
|
}
|
|
}
|
|
}
|
|
} catch (dbE) {
|
|
logger.warn(`[Executor] Could not recover state from DB for ${symbol}: ${dbE}`);
|
|
}
|
|
|
|
this.upsertPosition(symbol, {
|
|
symbol,
|
|
side: finalSide,
|
|
entryPrice: recoveredEntryPrice,
|
|
size: recoveredSize,
|
|
stopLoss: recoveredSl,
|
|
takeProfit: recoveredTp,
|
|
peakPrice: recoveredEntryPrice,
|
|
userId: recoveredUserId,
|
|
profileId: this.profileId,
|
|
tradeId: recoveredTradeId // Now persisted
|
|
});
|
|
} catch (e) {
|
|
logger.error(`[Executor] Sync failed for ${symbol}: ${e}`);
|
|
}
|
|
}
|
|
// --- NEW: Immediately sync to dashboard after syncing with exchange ---
|
|
if (this.apiServer) {
|
|
const allPos = this.getAllActivePositions();
|
|
this.apiServer.updatePositions(allPos, this.profileId || 'global');
|
|
this.broadcastOrders();
|
|
}
|
|
|
|
// --- NEW: Recover pending orders from DB for background sync ---
|
|
if (this.profileId) {
|
|
try {
|
|
const pending = await runtimeOrderRepository.getPendingOrdersForProfile(this.profileId);
|
|
for (const p of pending) {
|
|
const pendingOrderId = String(p.order_id || '').trim();
|
|
if (!pendingOrderId) {
|
|
continue;
|
|
}
|
|
const normalizedAction = String(p.action || '').toUpperCase();
|
|
const normalizedSide = normalizeTradeSide(p.side || 'BUY');
|
|
const isExitLike = normalizedAction === 'EXIT'
|
|
|| (!normalizedAction && normalizedSide === 'SELL');
|
|
|
|
if (isExitLike) {
|
|
let lifecycleClosed = false;
|
|
const normalizedTradeId = String(p.trade_id || '').trim();
|
|
|
|
if (normalizedTradeId) {
|
|
lifecycleClosed = await runtimeOrderRepository.isTradeLifecycleClosed(
|
|
normalizedTradeId,
|
|
this.profileId,
|
|
p.symbol
|
|
);
|
|
} else {
|
|
const virtualPosition = await runtimeOrderRepository.getVirtualOpenPosition(this.profileId, p.symbol);
|
|
lifecycleClosed = !virtualPosition;
|
|
}
|
|
|
|
if (lifecycleClosed) {
|
|
await runtimeOrderRepository.updateOrderStatus(pendingOrderId, 'canceled');
|
|
logger.warn(`[Executor] Auto-resolved stale pending EXIT ${pendingOrderId} for ${p.symbol} under profile ${this.profileId}.`);
|
|
continue;
|
|
}
|
|
}
|
|
|
|
if (!this.pendingOrders.has(pendingOrderId)) {
|
|
const normalizedSide = normalizeTradeSide(p.side || 'BUY');
|
|
const resolvedAction = normalizeOrderAction(p.action || undefined)
|
|
|| (normalizedSide === SignalDirection.SELL ? 'EXIT' : 'ENTRY');
|
|
this.pendingOrders.set(pendingOrderId, {
|
|
orderId: pendingOrderId,
|
|
symbol: p.symbol,
|
|
side: normalizedSide as SignalDirection,
|
|
qty: p.qty,
|
|
type: (p.type || 'market').toLowerCase() as 'market' | 'limit',
|
|
requestedPrice: p.price,
|
|
stopLoss: p.stop_loss || 0,
|
|
takeProfit: p.take_profit || 0,
|
|
tradeId: p.trade_id || '',
|
|
userId: p.user_id,
|
|
subTag: extractOrderSubTag(p) || undefined,
|
|
placedAt: new Date(p.created_at || Date.now()).getTime(),
|
|
action: resolvedAction
|
|
});
|
|
logger.info(`[Executor] 🔄 Recovered pending order ${p.order_id} for ${p.symbol} into monitoring map.`);
|
|
}
|
|
}
|
|
} catch (pE) {
|
|
logger.warn(`[Executor] Failed to recover pending orders: ${pE}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Returns all currently active trades for dashboard display
|
|
*/
|
|
public getAllActivePositions() {
|
|
return Array.from(this.activeTraders.values())
|
|
.filter((pos) => pos.side !== SignalDirection.NONE)
|
|
.map((pos) => ({
|
|
id: pos.tradeId || `pos-${pos.symbol}`,
|
|
symbol: pos.symbol,
|
|
side: pos.side as 'BUY' | 'SELL',
|
|
size: pos.size,
|
|
entryPrice: pos.entryPrice,
|
|
currentPrice: pos.peakPrice || pos.entryPrice,
|
|
stopLoss: pos.stopLoss,
|
|
takeProfit: pos.takeProfit,
|
|
unrealizedPnl: 0,
|
|
unrealizedPnlPercent: 0,
|
|
marketValue: pos.size * (pos.peakPrice || pos.entryPrice),
|
|
userId: pos.userId,
|
|
profileId: pos.profileId,
|
|
profileName: '', // Will be matched by dashboard or service
|
|
tradeId: pos.tradeId
|
|
}));
|
|
}
|
|
|
|
public getProfileId() {
|
|
return this.profileId;
|
|
}
|
|
|
|
public getUserId() {
|
|
return this.userId;
|
|
}
|
|
|
|
public async checkExchangeOrderStatus(orderId: string, symbol: string): Promise<string | null> {
|
|
if (!this.exchange.getOrder) return null;
|
|
try {
|
|
const tradeSymbol = SymbolMapper.toTradeSymbol(symbol, config.EXECUTION_PROVIDER);
|
|
const order = await this.exchange.getOrder(orderId, tradeSymbol);
|
|
const status = String(order?.status || '').trim().toLowerCase();
|
|
return status || null;
|
|
} catch (error: any) {
|
|
logger.debug(`[Executor] Exchange order status check failed for ${orderId}/${symbol}: ${error.message}`);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
private async instrumentExchangeCall<T>(operation: string, fn: () => Promise<T>): Promise<T> {
|
|
const start = Date.now();
|
|
try {
|
|
return await fn();
|
|
} catch (error: any) {
|
|
const errorMsg = error.message || String(error);
|
|
if (operation !== 'fetch_ohlcv' && operation !== 'get_position' && operation !== 'fetch_open_orders') {
|
|
observabilityService.emitEvent({
|
|
type: 'SYSTEM_ERROR',
|
|
severity: 'WARN',
|
|
message: `Exchange API ${operation} failed: ${errorMsg}`
|
|
});
|
|
}
|
|
throw error;
|
|
} finally {
|
|
observabilityService.observeExchangeLatency(operation, Date.now() - start);
|
|
}
|
|
}
|
|
}
|