Stan Runge
abac077541
All checks were successful
Test and deploy staging / staging (push) Successful in 11m52s
288 lines
10 KiB
TypeScript
288 lines
10 KiB
TypeScript
/**
|
|
* Steam Community Market ingestion. Run after db:migrate:cs-prices.
|
|
*
|
|
* bun scripts/ingest-cs-prices.ts catalog [--limit=500] [--start=0]
|
|
* STEAM_SESSIONID=... STEAM_LOGIN_SECURE=... bun scripts/ingest-cs-prices.ts history [--limit=100] [--start=0]
|
|
* bun scripts/ingest-cs-prices.ts import ./observations.jsonl
|
|
*
|
|
* The first two modes use Steam's public market web endpoints, which are not a
|
|
* contracted API and may rate limit or change. `import` remains provider-neutral.
|
|
*/
|
|
import { readFile } from 'node:fs/promises';
|
|
import postgres from 'postgres';
|
|
import { dimensionsFromName, type Dimensions } from '../src/lib/cs-prices/taxonomy';
|
|
import { steamDate } from '../src/lib/cs-prices/steam';
|
|
|
|
if (!process.env.DB_URL) throw new Error('DB_URL is required');
|
|
const sql = postgres(process.env.DB_URL, { max: 1 });
|
|
const mode = process.argv[2];
|
|
const flags = new Map(
|
|
process.argv
|
|
.slice(3)
|
|
.filter((arg) => arg.startsWith('--'))
|
|
.map((arg) => {
|
|
const [key, value] = arg.slice(2).split('=');
|
|
return [key, value ?? ''];
|
|
})
|
|
);
|
|
const start = Math.max(0, Number(flags.get('start') ?? 0));
|
|
const limit = Math.max(1, Number(flags.get('limit') ?? 1_000_000_000));
|
|
const delayMs = Math.max(1000, Number(flags.get('delay-ms') ?? 2000));
|
|
const sweepTime = new Date();
|
|
const knownStreams = new Set<string>();
|
|
|
|
type Observation = {
|
|
marketHashName: string;
|
|
name?: string;
|
|
source: string;
|
|
currency: string;
|
|
observedAt: string;
|
|
priceCents: number;
|
|
volume?: number | null;
|
|
kind: string;
|
|
dimensions?: Dimensions;
|
|
};
|
|
|
|
function validObservation(value: unknown): Observation {
|
|
if (!value || typeof value !== 'object') throw new Error('Observation must be an object');
|
|
const row = value as Record<string, unknown>;
|
|
for (const key of ['marketHashName', 'source', 'currency', 'observedAt', 'kind']) {
|
|
if (typeof row[key] !== 'string' || !row[key]) throw new Error(`Invalid ${key}`);
|
|
}
|
|
if (!/^[a-z0-9][a-z0-9-]{0,79}$/.test(row.source as string)) throw new Error('Invalid source');
|
|
if (!/^[a-z0-9][a-z0-9-]{0,79}$/.test(row.kind as string)) throw new Error('Invalid kind');
|
|
if (!/^[A-Z]{3}$/.test(row.currency as string)) throw new Error('Invalid currency');
|
|
if (row.name != null && typeof row.name !== 'string') throw new Error('Invalid name');
|
|
if (
|
|
!Number.isInteger(row.priceCents) ||
|
|
(row.priceCents as number) < 0 ||
|
|
(row.priceCents as number) > 2_147_483_647
|
|
)
|
|
throw new Error('Invalid priceCents');
|
|
if (
|
|
row.volume != null &&
|
|
(!Number.isInteger(row.volume) ||
|
|
(row.volume as number) < 0 ||
|
|
(row.volume as number) > 2_147_483_647)
|
|
)
|
|
throw new Error('Invalid volume');
|
|
if (Number.isNaN(Date.parse(row.observedAt as string))) throw new Error('Invalid observedAt');
|
|
if (
|
|
row.dimensions != null &&
|
|
(typeof row.dimensions !== 'object' ||
|
|
Array.isArray(row.dimensions) ||
|
|
Object.values(row.dimensions).some((value) => typeof value !== 'string'))
|
|
)
|
|
throw new Error('Invalid dimensions');
|
|
return row as Observation;
|
|
}
|
|
|
|
async function itemId(
|
|
name: string,
|
|
displayName = name,
|
|
iconUrl?: string,
|
|
itemType?: string,
|
|
extraDimensions: Dimensions = {}
|
|
) {
|
|
const [row] = await sql<{ id: number }[]>`
|
|
INSERT INTO cs_market_items (market_hash_name, name, icon_url, item_type)
|
|
VALUES (${name}, ${displayName}, ${iconUrl ?? null}, ${itemType ?? null})
|
|
ON CONFLICT (market_hash_name) DO UPDATE SET
|
|
name = EXCLUDED.name,
|
|
icon_url = COALESCE(EXCLUDED.icon_url, cs_market_items.icon_url),
|
|
item_type = COALESCE(EXCLUDED.item_type, cs_market_items.item_type),
|
|
last_seen_at = now()
|
|
RETURNING id`;
|
|
const inferred = dimensionsFromName(name);
|
|
if (itemType) inferred.steam_type = itemType;
|
|
for (const [key, value] of Object.entries(inferred)) {
|
|
if (!/^[a-z][a-z0-9_]*$/.test(key) || !value) continue;
|
|
await sql`
|
|
INSERT INTO cs_item_dimensions (item_id, key, value) VALUES (${row.id}, ${key}, ${value})
|
|
ON CONFLICT (item_id, key) DO NOTHING`;
|
|
}
|
|
for (const [key, value] of Object.entries(extraDimensions)) {
|
|
if (!/^[a-z][a-z0-9_]*$/.test(key) || !value) continue;
|
|
await sql`
|
|
INSERT INTO cs_item_dimensions (item_id, key, value) VALUES (${row.id}, ${key}, ${value})
|
|
ON CONFLICT (item_id, key) DO UPDATE SET value = EXCLUDED.value`;
|
|
}
|
|
return row.id;
|
|
}
|
|
|
|
async function saveObservation(row: Observation, knownItemId?: number) {
|
|
const item =
|
|
knownItemId ??
|
|
(await itemId(row.marketHashName, row.name, undefined, undefined, row.dimensions));
|
|
await ensureStream(row.source, row.currency, row.kind);
|
|
await sql`
|
|
INSERT INTO cs_price_observations
|
|
(item_id, source, currency, observed_at, price_cents, volume, kind)
|
|
VALUES (${item}, ${row.source}, ${row.currency}, ${new Date(row.observedAt)},
|
|
${row.priceCents}, ${row.volume ?? null}, ${row.kind})
|
|
ON CONFLICT (item_id, source, currency, observed_at, kind) DO UPDATE SET
|
|
price_cents = EXCLUDED.price_cents,
|
|
volume = EXCLUDED.volume`;
|
|
}
|
|
|
|
async function ensureStream(source: string, currency: string, kind: string) {
|
|
const key = `${source}\0${currency}\0${kind}`;
|
|
if (knownStreams.has(key)) return;
|
|
await sql`INSERT INTO cs_price_streams (source, currency, kind)
|
|
VALUES (${source}, ${currency}, ${kind}) ON CONFLICT DO NOTHING`;
|
|
knownStreams.add(key);
|
|
}
|
|
|
|
async function saveHistoryRows(
|
|
item: { id: number; market_hash_name: string },
|
|
prices: [string, number, string][]
|
|
) {
|
|
await ensureStream('steam-market', 'USD', 'sale-median');
|
|
const rows = prices
|
|
.filter(([, price]) => Number.isFinite(price) && price >= 0)
|
|
.map(([date, price, volume]) => ({
|
|
item_id: item.id,
|
|
source: 'steam-market',
|
|
currency: 'USD',
|
|
observed_at: new Date(steamDate(date)),
|
|
price_cents: Math.round(price * 100),
|
|
volume: Number(volume.replaceAll(',', '')) || 0,
|
|
kind: 'sale-median'
|
|
}));
|
|
for (let offset = 0; offset < rows.length; offset += 500) {
|
|
await sql`INSERT INTO cs_price_observations ${sql(rows.slice(offset, offset + 500))}
|
|
ON CONFLICT (item_id, source, currency, observed_at, kind) DO UPDATE SET
|
|
price_cents = EXCLUDED.price_cents, volume = EXCLUDED.volume`;
|
|
}
|
|
}
|
|
|
|
async function steamJson<T>(url: URL, authenticated = false): Promise<T> {
|
|
for (let attempt = 0; attempt < 5; attempt++) {
|
|
const cookie = authenticated
|
|
? `sessionid=${process.env.STEAM_SESSIONID ?? ''}; steamLoginSecure=${process.env.STEAM_LOGIN_SECURE ?? ''}`
|
|
: undefined;
|
|
const response = await fetch(url, {
|
|
headers: {
|
|
'User-Agent': 'cs-prices/1.0 personal historical archive',
|
|
...(cookie ? { Cookie: cookie } : {})
|
|
}
|
|
});
|
|
if (response.ok) {
|
|
const json = await response.json();
|
|
if (!json || json.success === false) throw new Error(`Steam rejected ${url.pathname}`);
|
|
return json as T;
|
|
}
|
|
if (response.status !== 429 && response.status < 500)
|
|
throw new Error(`Steam HTTP ${response.status}`);
|
|
await new Promise((resolve) => setTimeout(resolve, delayMs * 2 ** attempt));
|
|
}
|
|
throw new Error(`Steam request failed after retries: ${url.pathname}`);
|
|
}
|
|
|
|
const pause = () => new Promise((resolve) => setTimeout(resolve, delayMs));
|
|
|
|
async function catalog() {
|
|
let offset = start;
|
|
let processed = 0;
|
|
while (processed < limit) {
|
|
const url = new URL('https://steamcommunity.com/market/search/render/');
|
|
url.search = new URLSearchParams({
|
|
appid: '730',
|
|
currency: '1',
|
|
country: 'US',
|
|
norender: '1',
|
|
count: String(Math.min(100, limit - processed)),
|
|
start: String(offset),
|
|
sort_column: 'name',
|
|
sort_dir: 'asc'
|
|
}).toString();
|
|
const page = await steamJson<{
|
|
total_count: number;
|
|
results: {
|
|
hash_name: string;
|
|
name: string;
|
|
sell_price?: number;
|
|
sell_listings?: number;
|
|
asset_description?: { icon_url?: string; type?: string };
|
|
}[];
|
|
}>(url);
|
|
if (!page.results?.length) break;
|
|
for (const entry of page.results) {
|
|
if (!entry.hash_name) continue;
|
|
const id = await itemId(
|
|
entry.hash_name,
|
|
entry.name || entry.hash_name,
|
|
entry.asset_description?.icon_url,
|
|
entry.asset_description?.type
|
|
);
|
|
if (Number.isSafeInteger(entry.sell_price) && (entry.sell_price ?? -1) >= 0) {
|
|
await saveObservation(
|
|
{
|
|
marketHashName: entry.hash_name,
|
|
source: 'steam-market',
|
|
currency: 'USD',
|
|
observedAt: sweepTime.toISOString(),
|
|
priceCents: entry.sell_price!,
|
|
volume: entry.sell_listings,
|
|
kind: 'lowest-listing'
|
|
},
|
|
id
|
|
);
|
|
}
|
|
}
|
|
processed += page.results.length;
|
|
offset += page.results.length;
|
|
console.log(`Catalog: ${offset}/${page.total_count} listings`);
|
|
if (offset >= page.total_count) break;
|
|
await pause();
|
|
}
|
|
}
|
|
|
|
async function history() {
|
|
if (!process.env.STEAM_SESSIONID || !process.env.STEAM_LOGIN_SECURE)
|
|
throw new Error('STEAM_SESSIONID and STEAM_LOGIN_SECURE are required for Steam history');
|
|
const items = await sql<{ id: number; market_hash_name: string }[]>`
|
|
SELECT id, market_hash_name FROM cs_market_items ORDER BY id LIMIT ${limit} OFFSET ${start}`;
|
|
let done = 0;
|
|
for (const item of items) {
|
|
const url = new URL('https://steamcommunity.com/market/pricehistory/');
|
|
url.search = new URLSearchParams({
|
|
appid: '730',
|
|
market_hash_name: item.market_hash_name,
|
|
currency: '1',
|
|
country: 'US'
|
|
}).toString();
|
|
const data = await steamJson<{ prices: [string, number, string][] }>(url, true);
|
|
await saveHistoryRows(item, data.prices ?? []);
|
|
console.log(
|
|
`History: ${++done}/${items.length} ${item.market_hash_name} (${data.prices?.length ?? 0} points)`
|
|
);
|
|
await pause();
|
|
}
|
|
}
|
|
|
|
async function importFile(path: string) {
|
|
const lines = (await readFile(path, 'utf8')).split(/\r?\n/);
|
|
let imported = 0;
|
|
for (const [index, line] of lines.entries()) {
|
|
if (!line.trim()) continue;
|
|
try {
|
|
await saveObservation(validObservation(JSON.parse(line)));
|
|
imported++;
|
|
} catch (cause) {
|
|
throw new Error(`Invalid observation on line ${index + 1}: ${String(cause)}`, { cause });
|
|
}
|
|
}
|
|
console.log(`Imported ${imported} observations`);
|
|
}
|
|
|
|
try {
|
|
if (mode === 'catalog') await catalog();
|
|
else if (mode === 'history') await history();
|
|
else if (mode === 'import' && process.argv[3]) await importFile(process.argv[3]);
|
|
else throw new Error('Usage: ingest-cs-prices.ts catalog|history|import <file>');
|
|
await sql`ANALYZE cs_market_items, cs_item_dimensions, cs_price_observations`;
|
|
} finally {
|
|
await sql.end();
|
|
}
|