boki/stock-bot
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

fixed up prices, symbols / exchanges

Browse source
This commit is contained in:
Boki 2025-07-11 08:34:59 -04:00
parent d8e7449605
commit f196c5dcf4
9 changed files with 191 additions and 202 deletions
Download patch file Download diff file Expand all files Collapse all files

67
apps/stock/data-ingestion/src/handlers/eod/actions/corporate-actions.ts
View file

@ -1,14 +1,9 @@
import type { BaseHandler } from '@stock-bot/handlers';
import type { DataIngestionServices } from '../../../types';
import type { EodHandler } from '../eod.handler';
import { EOD_CONFIG } from '../shared';
import { getEodExchangeSuffix } from '../shared/utils';
interface FetchCorporateActionsInput {
symbol: string;
exchange: string;
eodSearchCode: string;
actionType: 'dividends' | 'splits';
country?: string;
}
export async function scheduleFetchCorporateActions(
@ -57,10 +52,8 @@ export async function scheduleFetchCorporateActions(
const { symbol } = staleSymbolsDividends[i];
await this.scheduleOperation('fetch-corporate-actions', {
symbol: symbol.Code,
exchange: symbol.eodExchange || symbol.Exchange, // Use eodExchange if available
actionType: 'dividends',
country: symbol.Country
eodSearchCode: symbol.eodSearchCode,
actionType: 'dividends'
}, {
attempts: 3,
backoff: {
@ -77,10 +70,8 @@ export async function scheduleFetchCorporateActions(
const { symbol } = staleSymbolsSplits[i];
await this.scheduleOperation('fetch-corporate-actions', {
symbol: symbol.Code,
exchange: symbol.eodExchange || symbol.Exchange, // Use eodExchange if available
actionType: 'splits',
country: symbol.Country
eodSearchCode: symbol.eodSearchCode,
actionType: 'splits'
}, {
attempts: 3,
backoff: {
@ -109,9 +100,27 @@ export async function fetchCorporateActions(
input: FetchCorporateActionsInput
): Promise<{ success: boolean; recordsCount: number }> {
const logger = this.logger;
const { symbol, exchange, actionType, country } = input;
const { eodSearchCode, actionType } = input;
// Declare variables for catch block
let symbol: string = '';
let exchange: string = '';
try {
// Lookup symbol using eodSearchCode
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
eodSearchCode: eodSearchCode
});
if (!symbolDoc) {
logger.error(`Symbol not found for eodSearchCode: ${eodSearchCode}`);
throw new Error(`Symbol not found: ${eodSearchCode}`);
}
symbol = symbolDoc.Code;
exchange = symbolDoc.eodExchange || symbolDoc.Exchange;
const country = symbolDoc.Country;
logger.info(`Fetching ${actionType} for ${symbol}.${exchange}`);
// Get API key
@ -120,23 +129,9 @@ export async function fetchCorporateActions(
throw new Error('EOD API key not configured');
}
// Get country if not provided
let symbolCountry = country;
if (!symbolCountry) {
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
Code: symbol,
Exchange: exchange
});
if (!symbolDoc) {
throw new Error(`Symbol ${symbol}.${exchange} not found in database`);
}
symbolCountry = symbolDoc.Country;
}
// Build URL based on action type
// Use utility function to handle US symbols and EUFUND special case
const exchangeSuffix = getEodExchangeSuffix(exchange, symbolCountry);
const exchangeSuffix = getEodExchangeSuffix(exchange, country);
const endpoint = actionType === 'dividends' ? 'div' : 'splits';
const url = new URL(`https://eodhd.com/api/${endpoint}/${symbol}.${exchangeSuffix}`);
@ -174,7 +169,7 @@ export async function fetchCorporateActions(
if (data.length === 0) {
// Update symbol to indicate we checked but found no data
await this.mongodb.collection('eodSymbols').updateOne(
{ Code: symbol, Exchange: exchange },
{ eodSearchCode: eodSearchCode },
{
$set: {
[`last${actionType.charAt(0).toUpperCase() + actionType.slice(1)}Update`]: new Date(),
@ -190,7 +185,7 @@ export async function fetchCorporateActions(
const recordsWithMetadata = data.map(record => ({
symbol,
exchange,
symbolExchange: `${symbol}.${exchange}`,
eodSearchCode,
...record,
actionType,
updatedAt: new Date(),
@ -200,16 +195,15 @@ export async function fetchCorporateActions(
// Determine collection name based on action type
const collectionName = actionType === 'dividends' ? 'eodDividends' : 'eodSplits';
// Save to MongoDB - use date and symbol as unique identifier
// Save to MongoDB - use date and eodSearchCode as unique identifier
const result = await this.mongodb.batchUpsert(
collectionName,
recordsWithMetadata,
['date', 'symbolExchange']
['date', 'eodSearchCode']
);
// Update operation tracker based on action type
const operationName = actionType === 'dividends' ? 'dividends_update' : 'splits_update';
const eodSearchCode = `${symbol}.${exchange}`;
await this.operationRegistry.updateOperation('eod', eodSearchCode, operationName, {
status: 'success',
recordCount: result.insertedCount,
@ -231,10 +225,9 @@ export async function fetchCorporateActions(
// Update operation tracker with failure
const operationName = actionType === 'dividends' ? 'dividends_update' : 'splits_update';
const eodSearchCode = `${symbol}.${exchange}`;
await this.operationRegistry.updateOperation('eod', eodSearchCode, operationName, {
status: 'failure',
error: error.message
error: error instanceof Error ? error.message : String(error)
});
throw error;

116
apps/stock/data-ingestion/src/handlers/eod/actions/fundamentals.ts
View file

@ -1,17 +1,11 @@
import type { BaseHandler } from '@stock-bot/handlers';
import type { DataIngestionServices } from '../../../types';
import type { EodHandler } from '../eod.handler';
import { EOD_CONFIG } from '../shared';
import { getEodExchangeSuffix } from '../shared/utils';
interface BulkFundamentalsInput {
symbols: Array<{ symbol: string; exchange: string; country?: string }>;
eodSearchCodes: string[];
}
interface FetchSingleFundamentalsInput {
symbol: string;
exchange: string;
country?: string;
eodSearchCode: string;
}
export async function scheduleFetchFundamentals(
@ -67,9 +61,7 @@ export async function scheduleFetchFundamentals(
for (let i = 0; i < etfs.length; i++) {
const { symbol: etf } = etfs[i];
await this.scheduleOperation('fetch-single-fundamentals', {
symbol: etf.Code,
exchange: etf.eodExchange || etf.Exchange, // Use eodExchange if available
country: etf.Country
eodSearchCode: etf.eodSearchCode
}, {
attempts: 3,
backoff: {
@ -91,15 +83,11 @@ export async function scheduleFetchFundamentals(
for (let i = 0; i < nonEtfs.length; i += batchSize) {
const batch = nonEtfs.slice(i, i + batchSize);
// Convert to array of {symbol, exchange, country} objects
const symbolBatch = batch.map(s => ({
symbol: s.symbol.Code,
exchange: s.symbol.eodExchange || s.symbol.Exchange, // Use eodExchange if available
country: s.symbol.Country
}));
// Convert to array of eodSearchCodes
const eodSearchCodes = batch.map(s => s.symbol.eodSearchCode);
await this.scheduleOperation('fetch-bulk-fundamentals', {
symbols: symbolBatch
eodSearchCodes: eodSearchCodes
}, {
attempts: 3,
backoff: {
@ -110,7 +98,7 @@ export async function scheduleFetchFundamentals(
});
jobsScheduled++;
logger.info(`Scheduled fundamentals batch with ${symbolBatch.length} non-ETF symbols`);
logger.info(`Scheduled fundamentals batch with ${eodSearchCodes.length} non-ETF symbols`);
}
logger.info(`Successfully scheduled ${jobsScheduled} fundamentals fetch jobs`);
@ -130,10 +118,20 @@ export async function fetchBulkFundamentals(
input: BulkFundamentalsInput
): Promise<{ success: boolean; symbolsProcessed: number }> {
const logger = this.logger;
const { symbols } = input;
const { eodSearchCodes } = input;
try {
logger.info('Fetching bulk fundamentals', { symbolCount: symbols.length });
logger.info('Fetching bulk fundamentals', { symbolCount: eodSearchCodes.length });
// Lookup all symbols
const symbolDocs = await this.mongodb.collection('eodSymbols').find({
eodSearchCode: { $in: eodSearchCodes }
}).toArray();
if (symbolDocs.length === 0) {
logger.error('No symbols found for provided eodSearchCodes');
return { success: true, symbolsProcessed: 0 };
}
// Get API key
const apiKey = EOD_CONFIG.API_TOKEN;
@ -142,7 +140,11 @@ export async function fetchBulkFundamentals(
}
// Group symbols by actual exchange for API endpoint, but use country for symbol suffix
const exchangeGroups = symbols.reduce((acc, { symbol, exchange, country }) => {
const exchangeGroups = symbolDocs.reduce((acc, symbolDoc) => {
const symbol = symbolDoc.Code;
const exchange = symbolDoc.eodExchange || symbolDoc.Exchange;
const country = symbolDoc.Country;
if (!acc[exchange]) {
acc[exchange] = [];
}
@ -193,20 +195,34 @@ export async function fetchBulkFundamentals(
}
// Extract symbol and exchange from the key
const [symbol, exc] = symbolExchange.split('.');
const [symbol, exchangeSuffix] = symbolExchange.split('.');
// Find the original symbol doc to get the actual exchange code
const symbolDoc = symbolDocs.find(doc =>
doc.Code === symbol &&
(doc.eodExchange === exchangeSuffix || doc.Exchange === exchangeSuffix)
);
if (!symbolDoc) {
logger.warn(`Could not find symbol doc for ${symbolExchange}`);
continue;
}
const actualExchange = symbolDoc.eodExchange || symbolDoc.Exchange;
const eodSearchCode = symbolDoc.eodSearchCode;
// Add metadata
const fundamentalsWithMetadata = {
symbol,
exchange: exc,
symbolExchange,
exchange: actualExchange,
eodSearchCode,
...fundamentals,
updatedAt: new Date(),
source: 'eod'
};
fundamentalsToSave.push(fundamentalsWithMetadata);
symbolsToUpdate.push({ symbol, exchange: exc });
symbolsToUpdate.push({ eodSearchCode });
}
if (fundamentalsToSave.length > 0) {
@ -214,14 +230,13 @@ export async function fetchBulkFundamentals(
const result = await this.mongodb.batchUpsert(
'eodFundamentals',
fundamentalsToSave,
['symbolExchange']
['eodSearchCode']
);
logger.info(`Saved ${result.insertedCount} fundamentals records for ${exchange}`);
// Update operation tracker for each symbol
const updatePromises = symbolsToUpdate.map(({ symbol, exchange }) => {
const eodSearchCode = `${symbol}.${exchange}`;
const updatePromises = symbolsToUpdate.map(({ eodSearchCode }) => {
return this.operationRegistry.updateOperation('eod', eodSearchCode, 'fundamentals_update', {
status: 'success',
recordCount: 1,
@ -247,8 +262,7 @@ export async function fetchBulkFundamentals(
logger.error('Failed to fetch bulk fundamentals', { error });
// Mark all symbols as failed
const failPromises = input.symbols.map(({ symbol, exchange }) => {
const eodSearchCode = `${symbol}.${exchange}`;
const failPromises = eodSearchCodes.map((eodSearchCode) => {
return this.operationRegistry.updateOperation('eod', eodSearchCode, 'fundamentals_update', {
status: 'failure',
error: error.message
@ -265,25 +279,29 @@ export async function fetchSingleFundamentals(
input: FetchSingleFundamentalsInput
): Promise<{ success: boolean; saved: boolean }> {
const logger = this.logger;
const { symbol, exchange, country } = input;
const { eodSearchCode } = input;
// Declare variables for catch block
let symbol: string = '';
let exchange: string = '';
try {
logger.info(`Fetching single fundamentals for ${symbol}.${exchange}`);
// Lookup symbol using eodSearchCode
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
eodSearchCode: eodSearchCode
});
// Get country if not provided
let symbolCountry = country;
if (!symbolCountry) {
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
Code: symbol,
Exchange: exchange
});
if (!symbolDoc) {
throw new Error(`Symbol ${symbol}.${exchange} not found in database`);
}
symbolCountry = symbolDoc.Country;
if (!symbolDoc) {
logger.error(`Symbol not found for eodSearchCode: ${eodSearchCode}`);
throw new Error(`Symbol not found: ${eodSearchCode}`);
}
symbol = symbolDoc.Code;
exchange = symbolDoc.eodExchange || symbolDoc.Exchange;
const country = symbolDoc.Country;
logger.info(`Fetching single fundamentals for ${symbol}.${exchange}`);
// Get API key
const apiKey = EOD_CONFIG.API_TOKEN;
if (!apiKey) {
@ -292,7 +310,7 @@ export async function fetchSingleFundamentals(
// Build URL for single fundamentals endpoint
// Use utility function to handle US symbols and EUFUND special case
const exchangeSuffix = getEodExchangeSuffix(exchange, symbolCountry);
const exchangeSuffix = getEodExchangeSuffix(exchange, country);
const url = new URL(`https://eodhd.com/api/fundamentals/${symbol}.${exchangeSuffix}`);
url.searchParams.append('api_token', apiKey);
@ -317,7 +335,7 @@ export async function fetchSingleFundamentals(
const fundamentalsWithMetadata = {
symbol,
exchange,
symbolExchange: `${symbol}.${exchange}`,
eodSearchCode,
...fundamentals,
updatedAt: new Date(),
source: 'eod'
@ -327,11 +345,10 @@ export async function fetchSingleFundamentals(
const result = await this.mongodb.batchUpsert(
'eodFundamentals',
[fundamentalsWithMetadata],
['symbolExchange']
['eodSearchCode']
);
// Update operation tracker
const eodSearchCode = `${symbol}.${exchange}`;
await this.operationRegistry.updateOperation('eod', eodSearchCode, 'fundamentals_update', {
status: 'success',
recordCount: result.insertedCount,
@ -352,7 +369,6 @@ export async function fetchSingleFundamentals(
logger.error('Failed to fetch single fundamentals', { error, symbol, exchange });
// Update operation tracker with failure
const eodSearchCode = `${symbol}.${exchange}`;
await this.operationRegistry.updateOperation('eod', eodSearchCode, 'fundamentals_update', {
status: 'failure',
error: error.message

98
apps/stock/data-ingestion/src/handlers/eod/actions/intraday.ts
View file

@ -3,19 +3,15 @@ import type { EodHandler } from '../eod.handler';
import { EOD_CONFIG } from '../shared';
interface FetchIntradayInput {
symbol: string;
exchange: string;
eodSearchCode: string;
interval: '1m' | '5m' | '1h';
fromDate?: Date;
toDate?: Date;
country?: string;
}
interface CrawlIntradayInput {
symbol: string;
exchange: string;
eodSearchCode: string;
interval: '1m' | '5m' | '1h';
country?: string;
}
@ -53,7 +49,7 @@ export async function scheduleIntradayCrawl(
}).toArray();
// Add interval info to each symbol
symbolsForInterval.forEach(symbol => {
symbolsForInterval.forEach((symbol: any) => {
// Check if this interval needs processing (not finished or needs new data)
const operationStatus = symbol.operations?.[operationName];
const shouldProcess = !operationStatus || !operationStatus.finished ||
@ -101,10 +97,8 @@ export async function scheduleIntradayCrawl(
const { symbol, interval } = item;
await this.scheduleOperation('crawl-intraday', {
symbol: symbol.Code,
exchange: symbol.eodExchange || symbol.Exchange, // Use eodExchange if available
interval,
country: symbol.Country
eodSearchCode: symbol.eodSearchCode,
interval
}, {
priority: 5, // Initial crawl jobs get priority 5 (lower priority)
attempts: 3,
@ -134,26 +128,31 @@ export async function crawlIntraday(
input: CrawlIntradayInput
): Promise<{ success: boolean; recordsProcessed: number; finished: boolean }> {
const logger = this.logger;
const { symbol, exchange, interval, country } = input;
const { eodSearchCode, interval } = input;
try {
// Lookup symbol using eodSearchCode
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
eodSearchCode: eodSearchCode
});
if (!symbolDoc) {
logger.error(`Symbol not found for eodSearchCode: ${eodSearchCode}`);
throw new Error(`Symbol not found: ${eodSearchCode}`);
}
const symbol = symbolDoc.Code;
const exchange = symbolDoc.eodExchange || symbolDoc.Exchange;
const country = symbolDoc.Country;
logger.info(`Starting intraday crawl for ${symbol}.${exchange} - ${interval}`, {
symbol,
exchange,
interval,
country
country,
eodSearchCode
});
// Get symbol to check if it exists
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
Code: symbol,
eodExchange: exchange
});
if (!symbolDoc) {
throw new Error(`Symbol ${symbol}.${exchange} not found`);
}
logger.debug('Found symbol document', {
symbol,
exchange,
@ -202,12 +201,10 @@ export async function crawlIntraday(
// Fetch data for this batch
const result = await fetchIntraday.call(this, {
symbol,
exchange,
eodSearchCode,
interval,
fromDate,
toDate,
country
toDate
});
// Prepare update data
@ -270,7 +267,6 @@ export async function crawlIntraday(
finished: updateData.finished
});
const eodSearchCode = `${symbol}.${exchange}`;
await this.operationRegistry.updateOperation('eod', eodSearchCode, operationName, updateData);
logger.info(`Operation tracker updated for ${symbol}.${exchange} - ${interval}`);
@ -278,10 +274,8 @@ export async function crawlIntraday(
// If not finished, schedule next batch
if (!updateData.finished) {
await this.scheduleOperation('crawl-intraday', {
symbol,
exchange,
interval,
country
eodSearchCode,
interval
}, {
priority: 3, // Continuation jobs get higher priority (3) than initial jobs (5)
attempts: 3,
@ -319,9 +313,27 @@ export async function fetchIntraday(
input: FetchIntradayInput
): Promise<{ success: boolean; recordsSaved: number; recordsFetched: number }> {
const logger = this.logger;
const { symbol, exchange, interval, fromDate, toDate, country } = input;
const { eodSearchCode, interval, fromDate, toDate } = input;
// Declare variables for catch block
let symbol: string = '';
let exchange: string = '';
try {
// Lookup symbol using eodSearchCode
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
eodSearchCode: eodSearchCode
});
if (!symbolDoc) {
logger.error(`Symbol not found for eodSearchCode: ${eodSearchCode}`);
throw new Error(`Symbol not found: ${eodSearchCode}`);
}
symbol = symbolDoc.Code;
exchange = symbolDoc.eodExchange || symbolDoc.Exchange;
const country = symbolDoc.Country;
logger.info(`Fetching intraday data for ${symbol}.${exchange} - ${interval}`, {
symbol,
exchange,
@ -332,20 +344,6 @@ export async function fetchIntraday(
url: `https://eodhd.com/api/intraday/${symbol}.${exchange}`
});
// Get country if not provided
let symbolCountry = country;
if (!symbolCountry) {
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
Code: symbol,
eodExchange: exchange
});
if (!symbolDoc) {
throw new Error(`Symbol ${symbol}.${exchange} not found in database`);
}
symbolCountry = symbolDoc.Country;
}
// Get API key
const apiKey = EOD_CONFIG.API_TOKEN;
if (!apiKey) {
@ -392,7 +390,7 @@ export async function fetchIntraday(
const recordsWithMetadata = data.map(bar => ({
symbol,
exchange,
symbolExchange: `${symbol}.${exchange}`,
eodSearchCode,
datetime: bar.datetime,
timestamp: bar.timestamp,
gmtoffset: bar.gmtoffset,
@ -404,12 +402,12 @@ export async function fetchIntraday(
source: 'eod'
}));
// Save to MongoDB - use timestamp and symbolExchange as unique identifier
// Save to MongoDB - use timestamp and eodSearchCode as unique identifier
const collectionName = `eodIntraday${interval.toUpperCase()}`;
const result = await this.mongodb.batchUpsert(
collectionName,
recordsWithMetadata,
['timestamp', 'symbolExchange']
['timestamp', 'eodSearchCode']
);
logger.info(`Saved ${result.insertedCount} intraday records`, {

73
apps/stock/data-ingestion/src/handlers/eod/actions/prices.ts
View file

@ -1,11 +1,8 @@
import type { EodHandler } from '../eod.handler';
import { EOD_CONFIG } from '../shared';
import { getEodExchangeSuffix } from '../shared/utils';
interface FetchPricesInput {
symbol: string;
exchange: string;
country?: string; // Optional to maintain backward compatibility
eodSearchCode: string;
}
export async function scheduleFetchPrices(
@ -18,7 +15,7 @@ export async function scheduleFetchPrices(
// Use OperationTracker to find stale symbols
const staleSymbols = await this.operationRegistry.getStaleSymbols('eod', 'price_update', {
limit: 50000 // Higher limit to process all symbols
limit: 14000 // Higher limit to process all symbols
});
if (!staleSymbols || staleSymbols.length === 0) {
@ -39,17 +36,20 @@ export async function scheduleFetchPrices(
// Schedule jobs with staggered delays
for (let i = 0; i < staleSymbols.length; i++) {
const { symbol } = staleSymbols[i];
const staleSymbol = staleSymbols[i];
if (!staleSymbol || !staleSymbol.symbol) {
logger.warn(`Skipping invalid stale symbol at index ${i}`, { staleSymbol });
continue;
}
const { symbol } = staleSymbol;
logger.debug(`Scheduling price fetch for ${symbol.Code}.${symbol.Exchange}`, {
name: symbol.Name,
lastUpdate: staleSymbols[i].lastRun,
lastUpdate: staleSymbol.lastRun,
delay: i * 100
});
await this.scheduleOperation('fetch-prices', {
symbol: symbol.Code,
exchange: symbol.eodExchange || symbol.Exchange, // Use eodExchange if available
country: symbol.Country
eodSearchCode: symbol.eodSearchCode
}, {
attempts: 3,
backoff: {
@ -78,36 +78,35 @@ export async function fetchPrices(
input: FetchPricesInput
): Promise<{ success: boolean; priceCount: number }> {
const logger = this.logger;
const { symbol, exchange, country } = input;
const { eodSearchCode } = input;
// Declare variables that need to be accessible in catch block
let symbol: string = '';
let exchange: string = '';
try {
logger.info(`Fetching prices for ${symbol}.${exchange}`);
// Lookup symbol using eodSearchCode
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
eodSearchCode: eodSearchCode
});
// Use provided country or fetch from database
let symbolCountry = country;
if (!symbolCountry) {
const symbolDoc = await this.mongodb.collection('eodSymbols').findOne({
Code: symbol,
Exchange: exchange
});
if (!symbolDoc) {
throw new Error(`Symbol ${symbol}.${exchange} not found in database`);
}
symbolCountry = symbolDoc.Country;
if (!symbolDoc) {
logger.error(`Symbol not found for eodSearchCode: ${eodSearchCode}`);
throw new Error(`Symbol not found: ${eodSearchCode}`);
}
symbol = symbolDoc.Code;
exchange = symbolDoc.Exchange;
logger.info(`Fetching prices for ${eodSearchCode}`);
// Get API key from config
const apiKey = EOD_CONFIG.API_TOKEN;
if (!apiKey) {
throw new Error('EOD API key not configured');
}
// Build URL for EOD price data
// Use utility function to handle US symbols and EUFUND special case
const exchangeSuffix = getEodExchangeSuffix(exchange, symbolCountry);
const url = new URL(`https://eodhd.com/api/eod/${symbol}.${exchangeSuffix}`);
const url = new URL(`https://eodhd.com/api/eod/${eodSearchCode}`);
url.searchParams.append('api_token', apiKey);
url.searchParams.append('fmt', 'json');
// Fetch price data from EOD API
@ -124,11 +123,11 @@ export async function fetchPrices(
throw new Error('Invalid response format from EOD API - expected array');
}
logger.info(`Fetched ${priceData.length} price records for ${symbol}.${exchange}`);
logger.info(`Fetched ${priceData.length} price records for ${eodSearchCode}`);
// Log date range of prices
if (priceData.length > 0) {
logger.debug(`Price data range for ${symbol}.${exchange}:`, {
logger.debug(`Price data range for ${eodSearchCode}:`, {
oldest: priceData[0].date,
newest: priceData[priceData.length - 1].date,
count: priceData.length
@ -139,7 +138,7 @@ export async function fetchPrices(
const pricesWithMetadata = priceData.map(price => ({
symbol,
exchange,
symbolExchange: `${symbol}.${exchange}`,
eodSearchCode,
date: price.date,
open: price.open,
high: price.high,
@ -149,15 +148,14 @@ export async function fetchPrices(
volume: price.volume,
}));
// Save to MongoDB - use date and symbol as unique identifier
// Save to MongoDB - use date and eodSearchCode as unique identifier
const result = await this.mongodb.batchUpsert(
'eodPrices',
pricesWithMetadata,
['date', 'symbolExchange']
['date', 'eodSearchCode']
);
// Update operation tracker instead of directly updating the symbol
const eodSearchCode = `${symbol}.${exchange}`;
await this.operationRegistry.updateOperation('eod', eodSearchCode, 'price_update', {
status: 'success',
lastRecordDate: priceData.length > 0 ? priceData[priceData.length - 1].date : null,
@ -168,7 +166,7 @@ export async function fetchPrices(
}
});
logger.info(`Successfully saved ${result.insertedCount} price records for ${symbol}.${exchange}`);
logger.info(`Successfully saved ${result.insertedCount} price records for ${eodSearchCode}`);
return {
success: true,
@ -178,10 +176,9 @@ export async function fetchPrices(
logger.error('Failed to fetch or save prices', { error, symbol, exchange });
// Update operation tracker with failure
const eodSearchCode = `${symbol}.${exchange}`;
await this.operationRegistry.updateOperation('eod', eodSearchCode, 'price_update', {
status: 'failure',
error: error.message
error: error instanceof Error ? error.message : String(error)
});
throw error;

7
apps/stock/data-ingestion/src/handlers/eod/actions/symbols.ts
View file

@ -25,7 +25,7 @@ export async function scheduleFetchSymbols(
}
logger.info(`Found ${exchanges.length} exchanges to process`, {
exchanges: exchanges.map(e => ({ code: e.Code, name: e.Name, country: e.Country }))
exchanges: exchanges.map((e: any) => ({ code: e.Code, name: e.Name, country: e.Country }))
});
let jobsCreated = 0;
@ -118,7 +118,7 @@ export async function fetchSymbols(
logger.debug(`Sample ${delisted ? 'delisted' : 'active'} symbols for ${exchangeCode}:`, {
count: symbols.length,
samples: symbols.slice(0, 5).map(s => ({
code: s.Code,
symbol: s.Code,
name: s.Name,
type: s.Type
}))
@ -128,7 +128,6 @@ export async function fetchSymbols(
// Add metadata to each symbol
const symbolsWithMetadata = symbols.map(symbol => ({
...symbol,
Exchange: symbol.Exchange || exchangeCode, // Keep the original exchange (might be wrong)
eodExchange: exchangeCode, // Store the correct exchange code used to fetch this symbol
eodSearchCode: `${symbol.Code}.${exchangeCode}`, // Create unique search code like AAPL.US
delisted: delisted,
@ -138,7 +137,7 @@ export async function fetchSymbols(
const result = await this.mongodb.batchUpsert(
'eodSymbols',
symbolsWithMetadata,
['Code', 'Exchange']
['eodSearchCode']
);
logger.info(`Successfully saved ${result.insertedCount} ${delisted ? 'delisted' : 'active'} symbols for ${exchangeCode}`);

2
apps/stock/data-ingestion/src/handlers/eod/eod.handler.ts
View file

@ -110,7 +110,7 @@ export class EodHandler extends BaseHandler<DataIngestionServices> {
*/
@Operation('schedule-intraday-crawl')
@ScheduledOperation('schedule-intraday-crawl', '0 3 * * *', {
immediately: true,
// immediately: true,
})
@RateLimit(1) // 1 point for scheduling
scheduleIntradayCrawl = scheduleIntradayCrawl;

2
apps/stock/data-ingestion/src/handlers/eod/shared/index.ts
View file

@ -1,3 +1,3 @@
export * from './config';
export * from './utils';
export * from './operation-provider';

21
apps/stock/data-ingestion/src/handlers/eod/shared/utils.ts
View file

@ -1,21 +0,0 @@
/**
* Get the exchange suffix for EOD API calls based on country and exchange
* US symbols use :US suffix, except EUFUND and GBOND which always use their own codes
* Others use their actual exchange code
*/
export function getEodExchangeSuffix(exchange: string, country?: string): string {
// Special cases that always use their own exchange code
if (exchange === 'EUFUND' || exchange === 'GBOND') {
return exchange;
}
// US symbols use :US suffix
return country === 'USA' ? 'US' : exchange;
}
/**
* Build symbol.exchange format for EOD API
*/
export function buildEodSymbol(symbol: string, exchange: string, country?: string): string {
const suffix = getEodExchangeSuffix(exchange, country);
return `${symbol}.${suffix}`;
}