refactoring to remove a lot of junk
This commit is contained in:
parent
5318158e59
commit
d858222af7
33 changed files with 505 additions and 367 deletions
3
apps/data-ingestion/src/handlers/ceo/actions/index.ts
Normal file
3
apps/data-ingestion/src/handlers/ceo/actions/index.ts
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
export { updateCeoChannels } from './update-ceo-channels.action';
|
||||
export { updateUniqueSymbols } from './update-unique-symbols.action';
|
||||
export { processIndividualSymbol } from './process-individual-symbol.action';
|
||||
|
|
@ -0,0 +1,111 @@
|
|||
import { getRandomUserAgent } from '@stock-bot/utils';
|
||||
import type { CeoHandler } from '../ceo.handler';
|
||||
|
||||
export async function processIndividualSymbol(this: CeoHandler, payload: any, _context: any): Promise<unknown> {
|
||||
const { ceoId, symbol, timestamp } = payload;
|
||||
const proxy = this.proxy?.getProxy();
|
||||
if(!proxy) {
|
||||
this.logger.warn('No proxy available for processing individual CEO symbol');
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.debug('Processing individual CEO symbol', {
|
||||
ceoId,
|
||||
timestamp,
|
||||
});
|
||||
try {
|
||||
// Fetch detailed information for the individual symbol
|
||||
const response = await this.http.get(`https://api.ceo.ca/api/get_spiels?channel=${ceoId}&load_more=top`
|
||||
+ (timestamp ? `&until=${timestamp}` : ''),
|
||||
{
|
||||
proxy: proxy,
|
||||
headers: {
|
||||
|
||||
'User-Agent': getRandomUserAgent()
|
||||
}
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to fetch details for ceoId ${ceoId}: ${response.statusText}`);
|
||||
}
|
||||
|
||||
const data = await response.json();
|
||||
|
||||
const spielCount = data.spiels.length;
|
||||
if(spielCount === 0) {
|
||||
this.logger.warn(`No spiels found for ceoId ${ceoId}`);
|
||||
return null; // No data to process
|
||||
}
|
||||
const latestSpielTime = data.spiels[0]?.timestamp;
|
||||
const posts = data.spiels.map((spiel: any) => ({
|
||||
ceoId,
|
||||
spiel: spiel.spiel,
|
||||
spielReplyToId: spiel.spiel_reply_to_id,
|
||||
spielReplyTo: spiel.spiel_reply_to,
|
||||
spielReplyToName: spiel.spiel_reply_to_name,
|
||||
spielReplyToEdited: spiel.spiel_reply_to_edited,
|
||||
userId: spiel.user_id,
|
||||
name: spiel.name,
|
||||
timestamp: spiel.timestamp,
|
||||
spielId: spiel.spiel_id,
|
||||
color: spiel.color,
|
||||
parentId: spiel.parent_id,
|
||||
publicId: spiel.public_id,
|
||||
parentChannel: spiel.parent_channel,
|
||||
parentTimestamp: spiel.parent_timestamp,
|
||||
votes: spiel.votes,
|
||||
editable: spiel.editable,
|
||||
edited: spiel.edited,
|
||||
featured: spiel.featured,
|
||||
verified: spiel.verified,
|
||||
fake: spiel.fake,
|
||||
bot: spiel.bot,
|
||||
voted: spiel.voted,
|
||||
flagged: spiel.flagged,
|
||||
ownSpiel: spiel.own_spiel,
|
||||
score: spiel.score,
|
||||
savedId: spiel.saved_id,
|
||||
savedTimestamp: spiel.saved_timestamp,
|
||||
poll: spiel.poll,
|
||||
votedInPoll: spiel.voted_in_poll
|
||||
}));
|
||||
|
||||
await this.mongodb.batchUpsert('ceoPosts', posts, ['spielId']);
|
||||
this.logger.info(`Fetched ${spielCount} spiels for ceoId ${ceoId}`);
|
||||
|
||||
// Update Shorts
|
||||
const shortRes = await this.http.get(`https://api.ceo.ca/api/short_positions/one?symbol=${symbol}`,
|
||||
{
|
||||
proxy: proxy,
|
||||
headers: {
|
||||
'User-Agent': getRandomUserAgent()
|
||||
}
|
||||
});
|
||||
|
||||
if (shortRes.ok) {
|
||||
const shortData = await shortRes.json();
|
||||
if(shortData && shortData.positions) {
|
||||
await this.mongodb.batchUpsert('ceoShorts', shortData.positions, ['id']);
|
||||
}
|
||||
|
||||
await this.scheduleOperation('process-individual-symbol', {
|
||||
ceoId: ceoId,
|
||||
timestamp: latestSpielTime
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
this.logger.info(`Successfully processed channel ${ceoId} and added channel ${ceoId} at timestamp ${latestSpielTime}`);
|
||||
|
||||
return { ceoId, spielCount, timestamp };
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to process individual symbol', {
|
||||
error,
|
||||
ceoId,
|
||||
timestamp
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,67 @@
|
|||
import { getRandomUserAgent } from '@stock-bot/utils';
|
||||
import type { CeoHandler } from '../ceo.handler';
|
||||
|
||||
export async function updateCeoChannels(this: CeoHandler, payload: number | undefined): Promise<unknown> {
|
||||
const proxy = this.proxy?.getProxy();
|
||||
if(!proxy) {
|
||||
this.logger.warn('No proxy available for CEO channels update');
|
||||
return;
|
||||
}
|
||||
let page;
|
||||
if(payload === undefined) {
|
||||
page = 1
|
||||
}else{
|
||||
page = payload;
|
||||
}
|
||||
|
||||
|
||||
this.logger.info(`Fetching CEO channels for page ${page} with proxy ${proxy}`);
|
||||
const response = await this.http.get('https://api.ceo.ca/api/home?exchange=all&sort_by=symbol§or=All&tab=companies&page='+page, {
|
||||
proxy: proxy,
|
||||
headers: {
|
||||
'User-Agent': getRandomUserAgent()
|
||||
}
|
||||
})
|
||||
const results = await response.json();
|
||||
const channels = results.channel_categories[0].channels;
|
||||
const totalChannels = results.channel_categories[0].total_channels;
|
||||
const totalPages = Math.ceil(totalChannels / channels.length);
|
||||
const exchanges: {exchange: string, countryCode: string}[] = []
|
||||
const symbols = channels.map((channel: any) =>{
|
||||
// check if exchange is in the exchanges array object
|
||||
if(!exchanges.find((e: any) => e.exchange === channel.exchange)) {
|
||||
exchanges.push({
|
||||
exchange: channel.exchange,
|
||||
countryCode: 'CA'
|
||||
});
|
||||
}
|
||||
const details = channel.company_details || {};
|
||||
return {
|
||||
symbol: channel.symbol,
|
||||
exchange: channel.exchange,
|
||||
name: channel.title,
|
||||
type: channel.type,
|
||||
ceoId: channel.channel,
|
||||
marketCap: details.market_cap,
|
||||
volumeRatio: details.volume_ratio,
|
||||
avgVolume: details.avg_volume,
|
||||
stockType: details.stock_type,
|
||||
issueType: details.issue_type,
|
||||
sharesOutstanding: details.shares_outstanding,
|
||||
float: details.float,
|
||||
}
|
||||
})
|
||||
|
||||
await this.mongodb.batchUpsert('ceoSymbols', symbols, ['symbol', 'exchange']);
|
||||
await this.mongodb.batchUpsert('ceoExchanges', exchanges, ['exchange']);
|
||||
|
||||
if(page === 1) {
|
||||
for( let i = 2; i <= totalPages; i++) {
|
||||
this.logger.info(`Scheduling page ${i} of ${totalPages} for CEO channels`);
|
||||
await this.scheduleOperation('update-ceo-channels', i)
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Fetched CEO channels for page ${page}/${totalPages}`);
|
||||
return { page, totalPages };
|
||||
}
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
import type { CeoHandler } from '../ceo.handler';
|
||||
|
||||
export async function updateUniqueSymbols(this: CeoHandler, _payload: unknown, _context: any): Promise<unknown> {
|
||||
this.logger.info('Starting update to get unique CEO symbols by ceoId');
|
||||
|
||||
try {
|
||||
// Get unique ceoId values from the ceoSymbols collection
|
||||
const uniqueCeoIds = await this.mongodb.collection('ceoSymbols').distinct('ceoId');
|
||||
|
||||
this.logger.info(`Found ${uniqueCeoIds.length} unique CEO IDs`);
|
||||
|
||||
// Get detailed records for each unique ceoId (latest/first record)
|
||||
const uniqueSymbols = [];
|
||||
for (const ceoId of uniqueCeoIds) {
|
||||
const symbol = await this.mongodb.collection('ceoSymbols')
|
||||
.findOne({ ceoId }, { sort: { _id: -1 } }); // Get latest record
|
||||
|
||||
if (symbol) {
|
||||
uniqueSymbols.push(symbol);
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Retrieved ${uniqueSymbols.length} unique symbol records`);
|
||||
|
||||
// Schedule individual jobs for each unique symbol
|
||||
let scheduledJobs = 0;
|
||||
for (const symbol of uniqueSymbols) {
|
||||
// Schedule a job to process this individual symbol
|
||||
await this.scheduleOperation('process-individual-symbol', {
|
||||
ceoId: symbol.ceoId,
|
||||
symbol: symbol.symbol,
|
||||
});
|
||||
scheduledJobs++;
|
||||
|
||||
// Add small delay to avoid overwhelming the queue
|
||||
if (scheduledJobs % 10 === 0) {
|
||||
this.logger.debug(`Scheduled ${scheduledJobs} jobs so far`);
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Successfully scheduled ${scheduledJobs} individual symbol update jobs`);
|
||||
|
||||
// Cache the results for monitoring
|
||||
await this.cacheSet('unique-symbols-last-run', {
|
||||
timestamp: new Date().toISOString(),
|
||||
totalUniqueIds: uniqueCeoIds.length,
|
||||
totalRecords: uniqueSymbols.length,
|
||||
scheduledJobs
|
||||
}, 1800); // Cache for 30 minutes
|
||||
|
||||
return {
|
||||
success: true,
|
||||
uniqueCeoIds: uniqueCeoIds.length,
|
||||
uniqueRecords: uniqueSymbols.length,
|
||||
scheduledJobs,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to update unique CEO symbols', { error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,15 +1,18 @@
|
|||
import {
|
||||
BaseHandler,
|
||||
Disabled,
|
||||
Handler,
|
||||
Operation,
|
||||
ScheduledOperation,
|
||||
type IServiceContainer
|
||||
} from '@stock-bot/handlers';
|
||||
import { getRandomUserAgent } from '@stock-bot/utils';
|
||||
import {
|
||||
processIndividualSymbol,
|
||||
updateCeoChannels,
|
||||
updateUniqueSymbols
|
||||
} from './actions';
|
||||
|
||||
@Handler('ceo')
|
||||
@Disabled()
|
||||
// @Disabled()
|
||||
export class CeoHandler extends BaseHandler {
|
||||
constructor(services: IServiceContainer) {
|
||||
super(services); // Handler name read from @Handler decorator
|
||||
|
|
@ -20,246 +23,16 @@ export class CeoHandler extends BaseHandler {
|
|||
immediately: false,
|
||||
description: 'Get all CEO symbols and exchanges'
|
||||
})
|
||||
async updateCeoChannels(payload: number | undefined): Promise<unknown> {
|
||||
const proxy = this.proxy?.getProxy();
|
||||
if(!proxy) {
|
||||
this.logger.warn('No proxy available for CEO channels update');
|
||||
return;
|
||||
}
|
||||
let page;
|
||||
if(payload === undefined) {
|
||||
page = 1
|
||||
}else{
|
||||
page = payload;
|
||||
}
|
||||
|
||||
|
||||
this.logger.info(`Fetching CEO channels for page ${page} with proxy ${proxy}`);
|
||||
const response = await this.http.get('https://api.ceo.ca/api/home?exchange=all&sort_by=symbol§or=All&tab=companies&page='+page, {
|
||||
proxy: proxy,
|
||||
headers: {
|
||||
'User-Agent': getRandomUserAgent()
|
||||
}
|
||||
})
|
||||
const results = await response.json();
|
||||
const channels = results.channel_categories[0].channels;
|
||||
const totalChannels = results.channel_categories[0].total_channels;
|
||||
const totalPages = Math.ceil(totalChannels / channels.length);
|
||||
const exchanges: {exchange: string, countryCode: string}[] = []
|
||||
const symbols = channels.map((channel: any) =>{
|
||||
// check if exchange is in the exchanges array object
|
||||
if(!exchanges.find((e: any) => e.exchange === channel.exchange)) {
|
||||
exchanges.push({
|
||||
exchange: channel.exchange,
|
||||
countryCode: 'CA'
|
||||
});
|
||||
}
|
||||
const details = channel.company_details || {};
|
||||
return {
|
||||
symbol: channel.symbol,
|
||||
exchange: channel.exchange,
|
||||
name: channel.title,
|
||||
type: channel.type,
|
||||
ceoId: channel.channel,
|
||||
marketCap: details.market_cap,
|
||||
volumeRatio: details.volume_ratio,
|
||||
avgVolume: details.avg_volume,
|
||||
stockType: details.stock_type,
|
||||
issueType: details.issue_type,
|
||||
sharesOutstanding: details.shares_outstanding,
|
||||
float: details.float,
|
||||
}
|
||||
})
|
||||
|
||||
await this.mongodb.batchUpsert('ceoSymbols', symbols, ['symbol', 'exchange']);
|
||||
await this.mongodb.batchUpsert('ceoExchanges', exchanges, ['exchange']);
|
||||
|
||||
if(page === 1) {
|
||||
for( let i = 2; i <= totalPages; i++) {
|
||||
this.logger.info(`Scheduling page ${i} of ${totalPages} for CEO channels`);
|
||||
await this.scheduleOperation('update-ceo-channels', i)
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Fetched CEO channels for page ${page}/${totalPages}`);
|
||||
return { page, totalPages };
|
||||
}
|
||||
updateCeoChannels = updateCeoChannels;
|
||||
|
||||
@Operation('update-unique-symbols')
|
||||
@ScheduledOperation('process-unique-symbols', '0 0 1 * *', {//'0 */30 * * *', {
|
||||
@ScheduledOperation('process-unique-symbols', '0 0 1 * *', {
|
||||
priority: 5,
|
||||
immediately: false,
|
||||
description: 'Process unique CEO symbols and schedule individual jobs'
|
||||
})
|
||||
async updateUniqueSymbols(_payload: unknown, _context: any): Promise<unknown> {
|
||||
this.logger.info('Starting update to get unique CEO symbols by ceoId');
|
||||
|
||||
try {
|
||||
// Get unique ceoId values from the ceoSymbols collection
|
||||
const uniqueCeoIds = await this.mongodb.collection('ceoSymbols').distinct('ceoId');
|
||||
|
||||
this.logger.info(`Found ${uniqueCeoIds.length} unique CEO IDs`);
|
||||
|
||||
// Get detailed records for each unique ceoId (latest/first record)
|
||||
const uniqueSymbols = [];
|
||||
for (const ceoId of uniqueCeoIds) {
|
||||
const symbol = await this.mongodb.collection('ceoSymbols')
|
||||
.findOne({ ceoId }, { sort: { _id: -1 } }); // Get latest record
|
||||
|
||||
if (symbol) {
|
||||
uniqueSymbols.push(symbol);
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Retrieved ${uniqueSymbols.length} unique symbol records`);
|
||||
|
||||
// Schedule individual jobs for each unique symbol
|
||||
let scheduledJobs = 0;
|
||||
for (const symbol of uniqueSymbols) {
|
||||
// Schedule a job to process this individual symbol
|
||||
await this.scheduleOperation('process-individual-symbol', {
|
||||
ceoId: symbol.ceoId,
|
||||
symbol: symbol.symbol,
|
||||
});
|
||||
scheduledJobs++;
|
||||
|
||||
// Add small delay to avoid overwhelming the queue
|
||||
if (scheduledJobs % 10 === 0) {
|
||||
this.logger.debug(`Scheduled ${scheduledJobs} jobs so far`);
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Successfully scheduled ${scheduledJobs} individual symbol update jobs`);
|
||||
|
||||
// Cache the results for monitoring
|
||||
await this.cacheSet('unique-symbols-last-run', {
|
||||
timestamp: new Date().toISOString(),
|
||||
totalUniqueIds: uniqueCeoIds.length,
|
||||
totalRecords: uniqueSymbols.length,
|
||||
scheduledJobs
|
||||
}, 1800); // Cache for 30 minutes
|
||||
|
||||
return {
|
||||
success: true,
|
||||
uniqueCeoIds: uniqueCeoIds.length,
|
||||
uniqueRecords: uniqueSymbols.length,
|
||||
scheduledJobs,
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to update unique CEO symbols', { error });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
updateUniqueSymbols = updateUniqueSymbols;
|
||||
|
||||
@Operation('process-individual-symbol')
|
||||
async processIndividualSymbol(payload: any, _context: any): Promise<unknown> {
|
||||
const { ceoId, symbol, timestamp } = payload;
|
||||
const proxy = this.proxy?.getProxy();
|
||||
if(!proxy) {
|
||||
this.logger.warn('No proxy available for processing individual CEO symbol');
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.debug('Processing individual CEO symbol', {
|
||||
ceoId,
|
||||
timestamp,
|
||||
});
|
||||
try {
|
||||
// Fetch detailed information for the individual symbol
|
||||
const response = await this.http.get(`https://api.ceo.ca/api/get_spiels?channel=${ceoId}&load_more=top`
|
||||
+ (timestamp ? `&until=${timestamp}` : ''),
|
||||
{
|
||||
proxy: proxy,
|
||||
headers: {
|
||||
|
||||
'User-Agent': getRandomUserAgent()
|
||||
}
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`Failed to fetch details for ceoId ${ceoId}: ${response.statusText}`);
|
||||
}
|
||||
|
||||
const data = await response.json();
|
||||
|
||||
const spielCount = data.spiels.length;
|
||||
if(spielCount === 0) {
|
||||
this.logger.warn(`No spiels found for ceoId ${ceoId}`);
|
||||
return null; // No data to process
|
||||
}
|
||||
const latestSpielTime = data.spiels[0]?.timestamp;
|
||||
const posts = data.spiels.map((spiel: any) => ({
|
||||
ceoId,
|
||||
spiel: spiel.spiel,
|
||||
spielReplyToId: spiel.spiel_reply_to_id,
|
||||
spielReplyTo: spiel.spiel_reply_to,
|
||||
spielReplyToName: spiel.spiel_reply_to_name,
|
||||
spielReplyToEdited: spiel.spiel_reply_to_edited,
|
||||
userId: spiel.user_id,
|
||||
name: spiel.name,
|
||||
timestamp: spiel.timestamp,
|
||||
spielId: spiel.spiel_id,
|
||||
color: spiel.color,
|
||||
parentId: spiel.parent_id,
|
||||
publicId: spiel.public_id,
|
||||
parentChannel: spiel.parent_channel,
|
||||
parentTimestamp: spiel.parent_timestamp,
|
||||
votes: spiel.votes,
|
||||
editable: spiel.editable,
|
||||
edited: spiel.edited,
|
||||
featured: spiel.featured,
|
||||
verified: spiel.verified,
|
||||
fake: spiel.fake,
|
||||
bot: spiel.bot,
|
||||
voted: spiel.voted,
|
||||
flagged: spiel.flagged,
|
||||
ownSpiel: spiel.own_spiel,
|
||||
score: spiel.score,
|
||||
savedId: spiel.saved_id,
|
||||
savedTimestamp: spiel.saved_timestamp,
|
||||
poll: spiel.poll,
|
||||
votedInPoll: spiel.voted_in_poll
|
||||
}));
|
||||
|
||||
await this.mongodb.batchUpsert('ceoPosts', posts, ['spielId']);
|
||||
this.logger.info(`Fetched ${spielCount} spiels for ceoId ${ceoId}`);
|
||||
|
||||
// Update Shorts
|
||||
const shortRes = await this.http.get(`https://api.ceo.ca/api/short_positions/one?symbol=${symbol}`,
|
||||
{
|
||||
proxy: proxy,
|
||||
headers: {
|
||||
'User-Agent': getRandomUserAgent()
|
||||
}
|
||||
});
|
||||
|
||||
if (shortRes.ok) {
|
||||
const shortData = await shortRes.json();
|
||||
if(shortData && shortData.positions) {
|
||||
await this.mongodb.batchUpsert('ceoShorts', shortData.positions, ['id']);
|
||||
}
|
||||
|
||||
await this.scheduleOperation('process-individual-symbol', {
|
||||
ceoId: ceoId,
|
||||
timestamp: latestSpielTime
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
this.logger.info(`Successfully processed channel ${ceoId} and added channel ${ceoId} at timestamp ${latestSpielTime}`);
|
||||
|
||||
return { ceoId, spielCount, timestamp };
|
||||
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to process individual symbol', {
|
||||
error,
|
||||
ceoId,
|
||||
timestamp
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
processIndividualSymbol = processIndividualSymbol;
|
||||
}
|
||||
|
|
@ -63,7 +63,7 @@ async function testCeoOperations() {
|
|||
if (count > 0) {
|
||||
// Test 2: Run process-unique-symbols operation
|
||||
logger.info('Testing process-unique-symbols operation...');
|
||||
const result = await ceoHandler.processUniqueSymbols(undefined, {});
|
||||
const result = await ceoHandler.updateUniqueSymbols(undefined, {});
|
||||
logger.info('Process unique symbols result:', result);
|
||||
|
||||
// Test 3: Test individual symbol processing
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue