cs-prices: price history backfill (Steam daily median since 2013 from the public listing page, CSFloat daily averages, ECB FX history) in separate *-history streams; labelled on item charts

2026-10-26 09:32:55 +00:00
parent 7396d23335
commit 76f063b362
5 changed files with 42972 additions and 2 deletions
--- a/drizzle/20261026091249_cs_history/migration.sql
+++ b/drizzle/20261026091249_cs_history/migration.sql
@@ -0,0 +1,11 @@
+CREATE TABLE "cs"."history_fetches" (
+	"item_id" integer,
+	"source" text,
+	"fetched_at" timestamp with time zone DEFAULT now() NOT NULL,
+	"points" integer DEFAULT 0 NOT NULL,
+	"first_day" date,
+	"error" text,
+	CONSTRAINT "history_fetches_pkey" PRIMARY KEY("item_id","source")
+);
+--> statement-breakpoint
+ALTER TABLE "cs"."history_fetches" ADD CONSTRAINT "history_fetches_item_id_items_id_fkey" FOREIGN KEY ("item_id") REFERENCES "cs"."items"("id") ON DELETE CASCADE;
--- a/drizzle/20261026091249_cs_history/snapshot.json
+++ b/drizzle/20261026091249_cs_history/snapshot.json
--- a/scripts/cs-history.ts
+++ b/scripts/cs-history.ts
@@ -0,0 +1,399 @@
+/**
+ * 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();
+}
--- a/src/lib/server/db/schema/cs-prices.ts
+++ b/src/lib/server/db/schema/cs-prices.ts
@@ -185,3 +185,21 @@ export const csFxRates = csSchema.table('fx_rates', {
 	day: date('day').primaryKey(),
 	usdEur: numeric('usd_eur', { precision: 12, scale: 8 }).notNull()
 });
+
+// Backfill bookkeeping (scripts/cs-history.ts): which config was fetched from which history
+// source, so the slow, throttled backfill can resume. History rows themselves live in
+// daily_prices under their own streams (kind ending in `-history`).
+export const csHistoryFetches = csSchema.table(
+	'history_fetches',
+	{
+		itemId: integer('item_id')
+			.notNull()
+			.references(() => csMarketItems.id, { onDelete: 'cascade' }),
+		source: text('source').notNull(),
+		fetchedAt: timestamp('fetched_at', { withTimezone: true }).notNull().defaultNow(),
+		points: integer('points').notNull().default(0),
+		firstDay: date('first_day'),
+		error: text('error')
+	},
+	(table) => [primaryKey({ name: 'history_fetches_pkey', columns: [table.itemId, table.source] })]
+);
--- a/src/routes/projects/cs-prices/[slug]/+page.svelte
+++ b/src/routes/projects/cs-prices/[slug]/+page.svelte
@@ -13,7 +13,9 @@
 		'steam-market:sale-avg': '#a855f7',
 		'steam-market:lowest-listing': '#d946ef',
 		'csfloat:lowest-listing': '#f59e0b',
-		'buff163:lowest-listing': '#ef4444'
+		'buff163:lowest-listing': '#ef4444',
+		'steam-market:sale-median-history': '#14b8a6',
+		'csfloat:sale-avg-history': '#f97316'
 	};
 	const sourceNames: Record<string, string> = {
 		skinport: 'Skinport',
@@ -24,7 +26,9 @@
 	const kindNames: Record<string, string> = {
 		'lowest-listing': 'lowest listing',
 		'sale-median-24h': 'sold, 24h median',
-		'sale-avg': 'sold, average'
+		'sale-avg': 'sold, average',
+		'sale-median-history': 'sold, daily median (history)',
+		'sale-avg-history': 'sold, daily average (history)'
 	};
 	const eur = new Intl.NumberFormat('nl-NL', { style: 'currency', currency: 'EUR' });
 	const usd = new Intl.NumberFormat('nl-NL', { style: 'currency', currency: 'USD' });