website/scripts/cs-history.ts

/**
 * CS2 price history backfill into cs.daily_prices, under separate `*-history` streams so it never
 * mixes with the daily scrape. Resumable (cs.history_fetches) and throttled; run in the background:
 *
 *   bun scripts/cs-history.ts fx              ECB USD→EUR for every day since 2013 (Frankfurter),
 *                                             weekends/holidays carry the last rate forward
 *   bun scripts/cs-history.ts steam [max]     Steam Community Market daily median sale price +
 *                                             volume since listing, read from the public market
 *                                             listing page (no login). One page holds every wear /
 *                                             StatTrak / Souvenir sibling, so it is ~1 request per
 *                                             skin, paced ~9 s apart with backoff on 429.
 *   bun scripts/cs-history.ts csfloat-queue   print CSFloat targets (NDJSON) for an off-box fetcher
 *   bun scripts/cs-history.ts csfloat-import  read CSFloat graph NDJSON on stdin and store it
 *   bun scripts/cs-history.ts status          coverage per history stream
 *
 * Steam prices arrive in the page's wallet currency (EUR here, `ecurrency` 3) and are stored as
 * is; USD streams (CSFloat) are converted per day with cs.fx_rates on read.
 */
import postgres from 'postgres';

if (!process.env.DB_URL) throw new Error('DB_URL is required');
const sql = postgres(process.env.DB_URL, { max: 2, onnotice: () => {} });
const mode = process.argv[2] ?? 'status';
const ua = 'Mozilla/5.0';
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
const log = (...args: unknown[]) => console.log(new Date().toISOString().slice(0, 19), ...args);

const streams = {
	steam: {
		source: 'steam-market',
		kind: 'sale-median-history',
		label: 'Steam Market daily median sale (history)'
	},
	csfloat: {
		source: 'csfloat',
		kind: 'sale-avg-history',
		label: 'CSFloat daily average sale (history)'
	}
};
async function streamId(source: string, currency: string, kind: string, label: string) {
	const [row] = await sql<{ id: number }[]>`
		INSERT INTO cs.price_streams (source, currency, kind, label)
		VALUES (${source}, ${currency}, ${kind}, ${label})
		ON CONFLICT (source, currency, kind) DO UPDATE SET label = EXCLUDED.label RETURNING id`;
	return row.id;
}

type Day = { day: string; cents: number; volume: number | null };
async function store(stream: number, item: number, days: Day[]) {
	for (let offset = 0; offset < days.length; offset += 5000) {
		const chunk = days.slice(offset, offset + 5000);
		await sql`
			INSERT INTO cs.daily_prices (item_id, stream_id, day, price_cents, volume)
			SELECT ${item}::int, ${stream}::smallint, day, price_cents, volume
			FROM unnest(${chunk.map((d) => d.day)}::date[], ${chunk.map((d) => d.cents)}::int[],
				${chunk.map((d) => d.volume)}::int[]) AS x(day, price_cents, volume)
			ON CONFLICT (item_id, stream_id, day) DO UPDATE SET
				price_cents = EXCLUDED.price_cents, volume = EXCLUDED.volume`;
	}
}
async function mark(item: number, source: string, days: Day[], error: string | null = null) {
	await sql`
		INSERT INTO cs.history_fetches (item_id, source, fetched_at, points, first_day, error)
		VALUES (${item}, ${source}, now(), ${days.length}, ${days[0]?.day ?? null}, ${error})
		ON CONFLICT (item_id, source) DO UPDATE SET fetched_at = now(), points = EXCLUDED.points,
			first_day = EXCLUDED.first_day, error = EXCLUDED.error`;
}

async function fx() {
	const from = '2013-01-01';
	const to = new Date().toISOString().slice(0, 10);
	let rates: Record<string, { EUR: number }> | undefined;
	for (const url of [
		`https://api.frankfurter.dev/v1/${from}..${to}?base=USD&symbols=EUR`,
		`https://api.frankfurter.app/${from}..${to}?from=USD&to=EUR`
	]) {
		const response = await fetch(url, { signal: AbortSignal.timeout(120_000) });
		if (response.ok) {
			rates = ((await response.json()) as { rates: typeof rates }).rates;
			break;
		}
	}
	if (!rates) throw new Error('No FX history');
	const days: string[] = [];
	const values: number[] = [];
	let last: number | null = null;
	for (let t = Date.parse(from); t <= Date.parse(to); t += 86_400_000) {
		const day = new Date(t).toISOString().slice(0, 10);
		last = rates[day]?.EUR ?? last;
		if (last != null) {
			days.push(day);
			values.push(last);
		}
	}
	// Existing rows (the daily scrape's own rate) win.
	const result = await sql`
		INSERT INTO cs.fx_rates (day, usd_eur)
		SELECT * FROM unnest(${days}::date[], ${values}::numeric[]) ON CONFLICT (day) DO NOTHING`;
	log(
		`FX: ${Object.keys(rates).length} ECB fixings, ${result.count} days inserted (${days[0]}..${to})`
	);
}

/** Every pricehistory query embedded in the listing page's SSR state. */
function pageHistories(html: string) {
	const match = html.match(
		/<script type="application\/json" id="valve-ssr-data"[^>]*>([\s\S]*?)<\/script>/
	);
	if (!match) return null;
	const found = new Map<
		string,
		{ currency: number; prices: { time: number; price_median: number; purchases: number }[] }
	>();
	const walk = (node: unknown): void => {
		if (typeof node === 'string') {
			if (node.length > 2 && (node[0] === '{' || node[0] === '[')) {
				try {
					walk(JSON.parse(node));
				} catch {
					/* not JSON */
				}
			}
			return;
		}
		if (Array.isArray(node)) return node.forEach(walk);
		if (node && typeof node === 'object') {
			const record = node as Record<string, unknown>;
			const key = record.queryKey;
			if (
				Array.isArray(key) &&
				key[0] === 'market' &&
				key[1] === 'pricehistory' &&
				key[2] === 730
			) {
				const data = (record.state as { data?: { ecurrency?: number; prices?: [] } } | undefined)
					?.data;
				if (data && Array.isArray(data.prices))
					found.set(String(key[3]), { currency: data.ecurrency ?? 0, prices: data.prices });
			}
			for (const value of Object.values(record)) walk(value);
		}
	};
	walk(JSON.parse(match[1]));
	return found;
}

/** Hourly points (last month) collapse to one row per UTC day, volume-weighted. */
function daily(prices: { time: number; price_median: number; purchases: number }[]): Day[] {
	const byDay = new Map<string, { sum: number; weight: number; volume: number }>();
	for (const p of prices) {
		if (!Number.isFinite(p.price_median) || p.price_median <= 0) continue;
		const day = new Date(p.time * 1000).toISOString().slice(0, 10);
		const weight = Math.max(1, p.purchases || 0);
		const entry = byDay.get(day) ?? { sum: 0, weight: 0, volume: 0 };
		entry.sum += p.price_median * weight;
		entry.weight += weight;
		entry.volume += p.purchases || 0;
		byDay.set(day, entry);
	}
	return [...byDay.entries()]
		.sort(([a], [b]) => a.localeCompare(b))
		.map(([day, e]) => ({ day, cents: Math.round((e.sum / e.weight) * 100), volume: e.volume }))
		.filter((d) => d.cents > 0 && d.cents < 2_000_000_000);
}

async function curlPage(url: string) {
	const proc = Bun.spawn(
		['curl', '-sL', '--compressed', '--max-time', '60', '-A', ua, '-w', '\n%{http_code}', url],
		{ stdout: 'pipe', stderr: 'ignore' }
	);
	const out = await new Response(proc.stdout).text();
	await proc.exited;
	const cut = out.lastIndexOf('\n');
	return { status: Number(out.slice(cut + 1)) || 0, body: out.slice(0, cut) };
}

const kindOrder = [
	'knife',
	'gloves',
	'weapon',
	'container',
	'agent',
	'key',
	'music-kit',
	'patch',
	'collectible',
	'sticker',
	'graffiti',
	'charm',
	'other'
];

async function steam(max: number) {
	const currencies: Record<number, string> = { 1: 'USD', 3: 'EUR' };
	const ids: Record<string, number> = {};
	const idFor = async (currency: string) =>
		(ids[currency] ??= await streamId(
			streams.steam.source,
			currency,
			streams.steam.kind,
			streams.steam.label
		));
	await idFor('EUR');
	// One representative unfetched phase-less config per skin, popular kinds/skins first.
	const queue = await sql<{ skin_id: string; item_id: number; hash: string }[]>`
		WITH todo AS (
			SELECT i.id, i.skin_id, i.market_hash_name, s.kind,
				coalesce((SELECT max(w.volume) FROM cs.price_windows w
					WHERE w.item_id = i.id AND w.source = 'skinport' AND w.span = '90d'), 0) AS vol
			FROM cs.items i JOIN cs.skins s ON s.id = i.skin_id
			WHERE i.phase IS NULL AND NOT EXISTS (SELECT 1 FROM cs.history_fetches f
				WHERE f.item_id = i.id AND f.source = 'steam')
		), skins AS (
			SELECT skin_id, kind, sum(vol) AS vol FROM todo GROUP BY 1, 2
		)
		SELECT DISTINCT ON (s.vol, s.skin_id) s.skin_id, t.id AS item_id, t.market_hash_name AS hash
		FROM skins s JOIN todo t ON t.skin_id = s.skin_id
		ORDER BY s.vol DESC, s.skin_id, t.vol DESC, t.id`;
	const kinds = new Map(
		(await sql<{ id: string; kind: string }[]>`SELECT id, kind FROM cs.skins`).map((r) => [
			r.id,
			r.kind
		])
	);
	const rank = (skin: string) => {
		const index = kindOrder.indexOf(kinds.get(skin) ?? 'other');
		return index < 0 ? kindOrder.length : index;
	};
	const ordered = queue
		.map((row, index) => ({ ...row, index }))
		.sort((a, b) => rank(a.skin_id) - rank(b.skin_id) || a.index - b.index);
	log(`Steam: ${ordered.length} skins with unfetched configs`);
	let delay = 9000;
	let strikes = 0;
	let requests = 0;
	let stored = 0;
	for (let i = 0; i < ordered.length; i++) {
		const row = ordered[i];
		if (requests >= max) break;
		// Skip when an earlier page already covered this skin's representative.
		const [done] =
			await sql`SELECT 1 FROM cs.history_fetches WHERE item_id = ${row.item_id} AND source = 'steam'`;
		if (done) continue;
		const siblings = await sql<{ id: number; hash: string }[]>`
			SELECT id, market_hash_name AS hash FROM cs.items WHERE skin_id = ${row.skin_id} AND phase IS NULL`;
		const byHash = new Map(siblings.map((s) => [s.hash, s.id]));
		const url = `https://steamcommunity.com/market/listings/730/${encodeURIComponent(row.hash)}`;
		requests++;
		// curl with a bare UA and no Accept-Language: Steam answers Bun's fetch, and any request
		// with Accept-Language, with 429 but serves this normally.
		let page: { status: number; body: string };
		try {
			page = await curlPage(url);
		} catch (cause) {
			log(`net error ${row.hash}: ${String(cause)}`);
			await sleep(60_000);
			i--; // retry the same skin
			continue;
		}
		if (page.status === 429 || page.status >= 500 || page.status === 0) {
			strikes++;
			delay = Math.min(30_000, delay + 1000);
			const pause = Math.min(30 * 60_000, 60_000 * strikes);
			log(
				`HTTP ${page.status} (strike ${strikes}), pausing ${pause / 1000}s, delay now ${delay / 1000}s`
			);
			await sleep(pause);
			i--; // retry the same skin
			continue;
		}
		strikes = 0;
		if (page.status !== 200) {
			await mark(row.item_id, 'steam', [], `HTTP ${page.status}`);
			await sleep(delay);
			continue;
		}
		const histories = pageHistories(page.body);
		if (!histories) {
			await mark(row.item_id, 'steam', [], 'no ssr data');
			log(`no SSR data for ${row.hash}`);
			await sleep(delay);
			continue;
		}
		let pageRows = 0;
		for (const [hash, history] of histories) {
			const item =
				byHash.get(hash) ??
				(
					await sql<{ id: number }[]>`SELECT id FROM cs.items WHERE market_hash_name = ${hash}
					AND phase IS NULL`
				)[0]?.id;
			if (!item) continue;
			const currency = currencies[history.currency];
			if (!currency) {
				await mark(item, 'steam', [], `currency ${history.currency}`);
				continue;
			}
			const days = daily(history.prices);
			if (days.length) await store(await idFor(currency), item, days);
			await mark(item, 'steam', days);
			pageRows += days.length;
		}
		if (!histories.has(row.hash)) await mark(row.item_id, 'steam', [], 'not on page');
		stored += pageRows;
		if (requests % 25 === 0 || requests <= 3)
			log(
				`#${requests} ${row.hash}: ${histories.size} configs, ${pageRows} days (total ${stored})`
			);
		await sleep(delay + Math.random() * 3000);
	}
	await sql`UPDATE cs.price_streams SET last_fetched_at = now(), last_rows = ${stored}
		WHERE source = ${streams.steam.source} AND kind = ${streams.steam.kind}`;
	log(`Steam done: ${requests} requests, ${stored} day rows`);
}

/** Targets for the CSFloat fetcher: phase-less configs plus Doppler phases (paint index added by
 * the fetcher), most traded first, skipping ones already fetched. */
async function csfloatQueue() {
	const rows = await sql<{ item_id: number; hash: string; phase: string | null; vol: number }[]>`
		SELECT i.id AS item_id, i.market_hash_name AS hash, i.phase,
			coalesce((SELECT max(w.volume) FROM cs.price_windows w WHERE w.item_id = i.id
				AND w.span = '90d'), 0)::int AS vol
		FROM cs.items i
		WHERE NOT EXISTS (SELECT 1 FROM cs.history_fetches f WHERE f.item_id = i.id AND f.source = 'csfloat')
		ORDER BY vol DESC, i.id`;
	for (const row of rows) process.stdout.write(JSON.stringify(row) + '\n');
}

async function csfloatImport() {
	const stream = await streamId(
		streams.csfloat.source,
		'USD',
		streams.csfloat.kind,
		streams.csfloat.label
	);
	let items = 0;
	let rows = 0;
	let buffer = '';
	const handle = async (line: string) => {
		if (!line.trim()) return;
		const entry = JSON.parse(line) as {
			item_id: number;
			error?: string;
			graph?: { day: string; avg_price: number; count: number }[];
		};
		const days = (entry.graph ?? [])
			.map((g) => ({
				day: g.day.slice(0, 10),
				cents: Math.round(g.avg_price),
				volume: g.count ?? null
			}))
			.filter((d) => d.cents > 0 && d.cents < 2_000_000_000)
			.sort((a, b) => a.day.localeCompare(b.day));
		if (days.length) await store(stream, entry.item_id, days);
		await mark(entry.item_id, 'csfloat', days, entry.error ?? null);
		items++;
		rows += days.length;
	};
	for await (const chunk of Bun.stdin.stream()) {
		buffer += new TextDecoder().decode(chunk);
		const lines = buffer.split('\n');
		buffer = lines.pop() ?? '';
		for (const line of lines) await handle(line);
	}
	await handle(buffer);
	await sql`UPDATE cs.price_streams SET last_fetched_at = now(), last_rows = ${rows} WHERE id = ${stream}`;
	log(`CSFloat import: ${items} configs, ${rows} day rows`);
}

async function status() {
	const rows = await sql`
		SELECT st.source, st.kind, st.currency, count(*)::int AS rows, count(DISTINCT dp.item_id)::int AS configs,
			min(dp.day)::text AS first_day, max(dp.day)::text AS last_day
		FROM cs.daily_prices dp JOIN cs.price_streams st ON st.id = dp.stream_id
		WHERE st.kind LIKE '%-history' GROUP BY 1, 2, 3 ORDER BY 1`;
	console.table(rows);
	const [cov] = await sql`
		WITH h AS (SELECT dp.item_id, count(DISTINCT dp.day) AS days FROM cs.daily_prices dp
			JOIN cs.price_streams st ON st.id = dp.stream_id WHERE st.kind LIKE '%-history' GROUP BY 1)
		SELECT (SELECT count(*) FROM cs.items)::int AS configs, count(*)::int AS with_history,
			count(*) FILTER (WHERE days > 30)::int AS over_30_days FROM h`;
	console.log(cov);
	console.table(
		await sql`SELECT source, count(*)::int AS fetched, sum(points)::int AS points,
		count(*) FILTER (WHERE points > 0)::int AS with_points, max(fetched_at) AS last
		FROM cs.history_fetches GROUP BY 1`
	);
}

try {
	if (mode === 'fx') await fx();
	else if (mode === 'steam') await steam(Number(process.argv[3] ?? Infinity));
	else if (mode === 'csfloat-queue') await csfloatQueue();
	else if (mode === 'csfloat-import') await csfloatImport();
	else if (mode === 'status') await status();
	else throw new Error('Usage: cs-history.ts fx|steam [max]|csfloat-queue|csfloat-import|status');
} finally {
	await sql.end();
}