Financial Time-Series Analytics

# tick_ingestion.py from cassandra.cluster import Cluster from cassandra.query import BatchStatement from datetime import datetime, date from decimal import Decimal class TickIngestion: def __init__(self, contact_points): cluster = Cluster(contact_points) self.session = cluster.connect('fintrack_data') # Prepared statements self.insert_tick = self.session.prepare(""" INSERT INTO ticks_by_symbol_day (symbol, day, tick_time, price, volume, bid, ask, exchange) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """) self.update_latest = self.session.prepare(""" UPDATE latest_prices SET last_price = ?, last_update = ?, day_high = ?, day_low = ?, day_volume = day_volume + ? WHERE symbol = ? """) def ingest_tick(self, tick): """Insert single tick and update latest price""" symbol = tick['symbol'] tick_time = tick['timestamp'] today = tick_time.date() # Insert tick data self.session.execute( self.insert_tick, ( symbol, today, tick_time, Decimal(str(tick['price'])), tick['volume'], Decimal(str(tick['bid'])), Decimal(str(tick['ask'])), tick['exchange'] ) ) # Update latest price (for real-time dashboard) self.session.execute( self.update_latest, ( Decimal(str(tick['price'])), tick_time, Decimal(str(tick['price'])), # Update high Decimal(str(tick['price'])), # Update low tick['volume'], symbol ) ) def ingest_batch(self, ticks): """Batch insert for high throughput""" batch = BatchStatement() for tick in ticks: today = tick['timestamp'].date() batch.add( self.insert_tick, ( tick['symbol'], today, tick['timestamp'], Decimal(str(tick['price'])), tick['volume'], Decimal(str(tick['bid'])), Decimal(str(tick['ask'])), tick['exchange'] ) ) self.session.execute(batch) # Usage ingestion = TickIngestion(['localhost']) # Real-time tick from market feed tick = { 'symbol': 'AAPL', 'timestamp': datetime.now(), 'price': 178.25, 'volume': 100, 'bid': 178.24, 'ask': 178.26, 'exchange': 'NASDAQ' } ingestion.ingest_tick(tick)

# market_queries.py from datetime import datetime, timedelta class MarketQueries: def __init__(self, contact_points): cluster = Cluster(contact_points) self.session = cluster.connect('fintrack_data') def get_latest_price(self, symbol): """Get current price""" query = "SELECT * FROM latest_prices WHERE symbol = ?" row = self.session.execute(query, (symbol,)).one() if row: return { 'symbol': row.symbol, 'price': float(row.last_price), 'change_percent': float(row.change_percent), 'volume': row.day_volume, 'high': float(row.day_high), 'low': float(row.day_low) } return None def get_intraday_candles(self, symbol, minutes=60): """Get 1-minute candles for last N minutes""" now = datetime.now() start_time = now - timedelta(minutes=minutes) # Calculate week bucket week_start = now - timedelta(days=now.weekday()) week = week_start.date() query = """ SELECT * FROM candles_1min WHERE symbol = ? AND week = ? AND candle_time >= ? AND candle_time <= ? """ rows = self.session.execute( query, (symbol, week, start_time, now) ) return [ { 'time': row.candle_time, 'open': float(row.open), 'high': float(row.high), 'low': float(row.low), 'close': float(row.close), 'volume': row.volume } for row in rows ] def get_daily_candles(self, symbol, start_date, end_date): """Get daily candles for date range""" # Query across multiple year partitions if needed years = range(start_date.year, end_date.year + 1) all_candles = [] for year in years: query = """ SELECT * FROM candles_daily WHERE symbol = ? AND year = ? AND candle_date >= ? AND candle_date <= ? """ rows = self.session.execute( query, (symbol, year, start_date, end_date) ) for row in rows: all_candles.append({ 'date': row.candle_date, 'open': float(row.open), 'high': float(row.high), 'low': float(row.low), 'close': float(row.close), 'volume': row.volume }) return sorted(all_candles, key=lambda x: x['date']) # Usage queries = MarketQueries(['localhost']) # Latest price price = queries.get_latest_price('AAPL') print(f"AAPL: ${price['price']} ({price['change_percent']}%)") # Intraday chart (last hour) candles = queries.get_intraday_candles('AAPL', minutes=60)

# candle_aggregator.py - Real-time aggregation from datetime import datetime, timedelta from decimal import Decimal class CandleAggregator: def __init__(self, session): self.session = session self.insert_1min = session.prepare(""" INSERT INTO candles_1min (symbol, week, candle_time, open, high, low, close, volume, trades, vwap) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """) def aggregate_1min_candle(self, symbol, candle_start): """Aggregate 1-minute candle from tick data""" candle_end = candle_start + timedelta(minutes=1) day = candle_start.date() # Get all ticks in 1-minute window query = """ SELECT price, volume FROM ticks_by_symbol_day WHERE symbol = ? AND day = ? AND tick_time >= ? AND tick_time < ? """ rows = self.session.execute( query, (symbol, day, candle_start, candle_end) ) ticks = list(rows) if not ticks: return # Calculate OHLC prices = [float(t.price) for t in ticks] volumes = [t.volume for t in ticks] ohlc = { 'open': prices[0], # First tick 'high': max(prices), 'low': min(prices), 'close': prices[-1], # Last tick 'volume': sum(volumes), 'trades': len(ticks) } # Calculate VWAP (Volume-Weighted Average Price) total_value = sum(p * v for p, v in zip(prices, volumes)) ohlc['vwap'] = total_value / ohlc['volume'] if ohlc['volume'] > 0 else 0 # Calculate week bucket week_start = candle_start - timedelta(days=candle_start.weekday()) week = week_start.date() # Insert candle self.session.execute( self.insert_1min, ( symbol, week, candle_start, Decimal(str(ohlc['open'])), Decimal(str(ohlc['high'])), Decimal(str(ohlc['low'])), Decimal(str(ohlc['close'])), ohlc['volume'], ohlc['trades'], Decimal(str(ohlc['vwap'])) ) ) return ohlc def run_aggregation_job(self, symbols): """Run every minute to aggregate previous minute""" now = datetime.now() # Round down to last complete minute candle_start = now.replace(second=0, microsecond=0) - timedelta(minutes=1) for symbol in symbols: self.aggregate_1min_candle(symbol, candle_start) print(f"Aggregated {len(symbols)} symbols for {candle_start}") # Schedule to run every minute import schedule aggregator = CandleAggregator(session) symbols = ['AAPL', 'GOOGL', 'MSFT', 'TSLA', ...] schedule.every().minute.at(":05").do(aggregator.run_aggregation_job, symbols)