103 lines
5.0 KiB
JavaScript
103 lines
5.0 KiB
JavaScript
const os = require('os');
|
|
const Database = require('better-sqlite3');
|
|
const { initAutonomySchema } = require('../src/autonomy/schema');
|
|
const { createOrderIntent } = require('../src/autonomy/orderIntents');
|
|
const { createAlpacaPaperClient } = require('../src/brokers/alpacaPaper');
|
|
|
|
function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
|
|
|
|
async function runExecutionWorker({ intelligencePath, pollMs = 10000, mode = 'shadow', notional = 100, workerId = `execution-${os.hostname()}-${process.pid}` } = {}) {
|
|
if (!['shadow', 'paper'].includes(mode)) throw new Error(`unsupported execution mode: ${mode}`);
|
|
const db = new Database(intelligencePath);
|
|
db.pragma('journal_mode = WAL');
|
|
db.pragma('busy_timeout = 5000');
|
|
initAutonomySchema(db);
|
|
const paperClient = mode === 'paper'
|
|
? createAlpacaPaperClient({ keyId: process.env.ALPACA_PAPER_KEY_ID, secretKey: process.env.ALPACA_PAPER_SECRET_KEY })
|
|
: null;
|
|
while (true) {
|
|
if (paperClient) {
|
|
try {
|
|
const [account, positions, orders] = await Promise.all([
|
|
paperClient.getAccount(), paperClient.getPositions(), paperClient.getOrders(),
|
|
]);
|
|
db.prepare(`
|
|
INSERT INTO autonomy_account_snapshots(broker, account_id, equity, cash, buying_power, payload)
|
|
VALUES ('alpaca-paper', ?, ?, ?, ?, ?)
|
|
`).run(account.id || null, Number(account.equity), Number(account.cash), Number(account.buying_power), JSON.stringify(account));
|
|
const insertPosition = db.prepare(`
|
|
INSERT INTO autonomy_position_snapshots(broker, instrument, quantity, market_value, unrealized_pl, payload)
|
|
VALUES ('alpaca-paper', ?, ?, ?, ?, ?)
|
|
`);
|
|
for (const position of positions || []) {
|
|
insertPosition.run(position.symbol, Number(position.qty), Number(position.market_value), Number(position.unrealized_pl), JSON.stringify(position));
|
|
}
|
|
for (const order of orders || []) {
|
|
if (!order.client_order_id) continue;
|
|
const mapped = { accepted: 'submitted', new: 'submitted', pending_new: 'submitted', partially_filled: 'partially_filled', filled: 'filled', canceled: 'cancelled', cancelled: 'cancelled', rejected: 'rejected' }[order.status];
|
|
if (!mapped) continue;
|
|
db.prepare(`
|
|
UPDATE autonomy_order_intents SET status=?, broker_order_id=?, updated_at=datetime('now')
|
|
WHERE client_order_id=?
|
|
`).run(mapped, order.id || null, order.client_order_id);
|
|
db.prepare(`
|
|
INSERT INTO autonomy_broker_events(broker, event_type, broker_id, payload)
|
|
VALUES ('alpaca-paper', ?, ?, ?)
|
|
`).run(order.status, order.id || null, JSON.stringify(order));
|
|
}
|
|
} catch (error) {
|
|
console.error(`[${workerId}] broker reconciliation:`, error.message);
|
|
}
|
|
}
|
|
const decisions = db.prepare(`
|
|
SELECT d.id FROM autonomy_decisions d
|
|
JOIN autonomy_predictions p ON p.id = d.prediction_id
|
|
LEFT JOIN autonomy_order_intents oi ON oi.decision_id = d.id
|
|
WHERE oi.id IS NULL AND d.action IN ('BUY', 'SELL') AND p.origin = 'live'
|
|
ORDER BY d.created_at ASC LIMIT 25
|
|
`).all();
|
|
for (const decision of decisions) {
|
|
try {
|
|
const intent = createOrderIntent(db, decision.id, notional, { tradable: true, maxNotional: notional });
|
|
if (mode === 'paper') db.prepare("UPDATE autonomy_order_intents SET status='pending', updated_at=datetime('now') WHERE client_order_id=?").run(intent.clientOrderId);
|
|
console.log(`[${workerId}] ${mode} intent ${intent.clientOrderId}`);
|
|
} catch (error) {
|
|
console.error(`[${workerId}] decision ${decision.id}:`, error.message);
|
|
}
|
|
}
|
|
if (paperClient) {
|
|
const pending = db.prepare("SELECT * FROM autonomy_order_intents WHERE status='pending' ORDER BY created_at ASC LIMIT 25").all();
|
|
for (const intent of pending) {
|
|
try {
|
|
let order;
|
|
try { order = await paperClient.getOrderByClientId(intent.client_order_id); } catch (error) {
|
|
if (error.status !== 404) throw error;
|
|
}
|
|
if (!order) {
|
|
order = await paperClient.submitOrder({
|
|
symbol: intent.instrument,
|
|
notional: String(intent.notional),
|
|
side: intent.side,
|
|
type: 'market',
|
|
time_in_force: 'day',
|
|
client_order_id: intent.client_order_id,
|
|
});
|
|
}
|
|
db.prepare(`
|
|
UPDATE autonomy_order_intents
|
|
SET status = ?, broker_order_id = ?, updated_at = datetime('now')
|
|
WHERE id = ?
|
|
`).run(order.status === 'filled' ? 'filled' : 'submitted', order.id || null, intent.id);
|
|
} catch (error) {
|
|
console.error(`[${workerId}] paper order ${intent.client_order_id}:`, error.message);
|
|
db.prepare("UPDATE autonomy_order_intents SET attempts=attempts+1, last_error=?, updated_at=datetime('now') WHERE id=?")
|
|
.run(String(error.message).slice(0, 1000), intent.id);
|
|
}
|
|
}
|
|
}
|
|
await sleep(pollMs);
|
|
}
|
|
}
|
|
|
|
module.exports = { runExecutionWorker };
|