#!/usr/bin/env python3 """ OHLCV Table Compression via Create-Copy-Swap strategy. Disk hanya 10GB free, ALTER TABLE butuh 16GB temp. Strategy: Create new compressed table → batch insert per timeframe → atomic rename. Ukuran compressed ~6-8 GB, cukup di 10 GB free space. """ import os, sys, time os.environ.setdefault('FLASK_ENV', 'development') # Add project root to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app import create_app from app.extensions import db from sqlalchemy import text SCHEMA = 'cornbyt1_invkamu_market' OLD_TABLE = 'ohlcv_data' NEW_TABLE = 'ohlcv_data_compressed' BATCH_SIZE = 50000 def fmt_mb(mb): return f"{mb:,.0f} MB ({mb/1024:.1f} GB)" if mb > 1024 else f"{mb:,.0f} MB" def get_table_size(table_name): r = db.session.execute(text(f""" SELECT ROUND(DATA_LENGTH/1024/1024,1), ROUND(INDEX_LENGTH/1024/1024,1), TABLE_ROWS FROM information_schema.TABLES WHERE TABLE_SCHEMA='{SCHEMA}' AND TABLE_NAME='{table_name}' """)).fetchone() return (float(r[0]), float(r[1]), int(r[2])) if r else (0, 0, 0) def get_disk_free(): """Get free disk space in MB.""" import shutil usage = shutil.disk_usage('/Applications/XAMPP/xamppfiles/var/mysql/') return usage.free / 1024 / 1024 def main(): app = create_app() with app.app_context(): print("=" * 60) print("OHLCV COMPRESSION: Create-Copy-Swap Strategy") print("=" * 60) # Pre-check disk_free = get_disk_free() print(f"\nDisk free: {fmt_mb(disk_free)}") if disk_free < 8000: print("⚠️ WARNING: Less than 8 GB free. Compressed table ~6-8 GB.") print(" Proceeding anyway, will monitor per timeframe.") data_mb, idx_mb, rows = get_table_size(OLD_TABLE) print(f"Source: {rows:,} rows | data={fmt_mb(data_mb)} idx={fmt_mb(idx_mb)}") # Step 1: Drop new table if exists (from failed previous run) print("\n[1/5] Preparing new compressed table...") db.session.execute(text(f"DROP TABLE IF EXISTS {SCHEMA}.{NEW_TABLE}")) db.session.commit() # Step 2: Create new table with compressed format # Use CREATE TABLE LIKE + ALTER (preserves structure including FK) db.session.execute(text(f""" CREATE TABLE {SCHEMA}.{NEW_TABLE} LIKE {SCHEMA}.{OLD_TABLE} """)) db.session.commit() db.session.execute(text(f""" ALTER TABLE {SCHEMA}.{NEW_TABLE} ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=8 """)) db.session.commit() print(" Created compressed table structure ✓") # Step 3: Copy data per timeframe (smallest first to detect issues early) print("\n[2/5] Copying data per timeframe...") timeframes_order = ['1W', '1D', '30m', '15m', '1m', '4h', '1h'] # smallest first # Get actual counts per timeframe tf_counts = db.session.execute(text(f""" SELECT timeframe, COUNT(*) FROM {SCHEMA}.{OLD_TABLE} GROUP BY timeframe ORDER BY COUNT(*) """)).fetchall() tf_map = {tf: cnt for tf, cnt in tf_counts} total_copied = 0 t_start = time.time() for tf in timeframes_order: if tf not in tf_map: continue expected = tf_map[tf] print(f"\n Timeframe {tf}: {expected:,} rows", end="", flush=True) tf_t0 = time.time() # Batch insert using INSERT ... SELECT with LIMIT/OFFSET # More efficient: insert all at once per timeframe db.session.execute(text(f""" INSERT INTO {SCHEMA}.{NEW_TABLE} SELECT * FROM {SCHEMA}.{OLD_TABLE} WHERE timeframe = :tf """), {'tf': tf}) db.session.commit() tf_elapsed = time.time() - tf_t0 total_copied += expected # Verify count new_count = db.session.execute(text(f""" SELECT COUNT(*) FROM {SCHEMA}.{NEW_TABLE} WHERE timeframe = :tf """), {'tf': tf}).scalar() status = "✓" if new_count == expected else f"⚠️ {new_count} vs {expected}" speed = expected / tf_elapsed if tf_elapsed > 0 else 0 print(f" → {status} ({tf_elapsed:.0f}s, {speed:,.0f} rows/s)") # Check disk space after each timeframe disk_now = get_disk_free() new_data, new_idx, _ = get_table_size(NEW_TABLE) print(f" New table: {fmt_mb(new_data + new_idx)} | Disk free: {fmt_mb(disk_now)}") if disk_now < 1000: print("\n❌ ABORT: Less than 1 GB disk free!") db.session.execute(text(f"DROP TABLE IF EXISTS {SCHEMA}.{NEW_TABLE}")) db.session.commit() print(" Cleaned up new table.") return total_elapsed = time.time() - t_start print(f"\n Total copied: {total_copied:,} rows in {total_elapsed:.0f}s ({total_elapsed/60:.1f} min)") # Step 4: Verify counts print("\n[3/5] Verifying row counts...") old_count = db.session.execute(text(f"SELECT COUNT(*) FROM {SCHEMA}.{OLD_TABLE}")).scalar() new_count = db.session.execute(text(f"SELECT COUNT(*) FROM {SCHEMA}.{NEW_TABLE}")).scalar() print(f" Old table: {old_count:,}") print(f" New table: {new_count:,}") if old_count != new_count: print(f"\n❌ COUNT MISMATCH! Aborting. Not dropping new table for inspection.") return print(" Counts match ✓") # Fix auto_increment max_id = db.session.execute(text(f"SELECT MAX(id) FROM {SCHEMA}.{NEW_TABLE}")).scalar() db.session.execute(text(f"ALTER TABLE {SCHEMA}.{NEW_TABLE} AUTO_INCREMENT = {max_id + 1}")) db.session.commit() # Step 5: Atomic rename print("\n[4/5] Atomic rename (old → backup, new → live)...") db.session.execute(text(f""" RENAME TABLE {SCHEMA}.{OLD_TABLE} TO {SCHEMA}.ohlcv_data_old, {SCHEMA}.{NEW_TABLE} TO {SCHEMA}.{OLD_TABLE} """)) db.session.commit() print(" Renamed ✓") # Step 6: Drop old table (frees ~16 GB!) print("\n[5/5] Dropping old uncompressed table (frees ~16 GB)...") db.session.execute(text(f"DROP TABLE {SCHEMA}.ohlcv_data_old")) db.session.commit() print(" Dropped ✓") # Final stats print("\n" + "=" * 60) print("RESULT") print("=" * 60) new_data, new_idx, new_rows = get_table_size(OLD_TABLE) disk_final = get_disk_free() print(f" Rows: {new_rows:,}") print(f" Data: {fmt_mb(new_data)}") print(f" Index: {fmt_mb(new_idx)}") print(f" Total: {fmt_mb(new_data + new_idx)}") print(f" Saved: {fmt_mb((data_mb + idx_mb) - (new_data + new_idx))}") print(f" Disk free: {fmt_mb(disk_final)}") print(f" Compression: {(1 - (new_data + new_idx)/(data_mb + idx_mb))*100:.0f}%") if __name__ == '__main__': main()