"""Sync market data (profiles + OHLCV) dari terminal — cocok untuk cron job. Sync semua coin aktif (kripto / saham IDX / saham US) dari provider (CoinGecko, Indodax, Yahoo Finance) ke database lokal. Usage: # Sync semua kripto (default mode) python3 scripts/sync_data.py # Sync saham IDX saja python3 scripts/sync_data.py --mode stock # Sync saham US saja python3 scripts/sync_data.py --mode stock_us # Sync semua mode sekaligus python3 scripts/sync_data.py --mode all # Hanya OHLCV (skip profile & icon) python3 scripts/sync_data.py --ohlcv-only # Hanya profile (skip OHLCV) python3 scripts/sync_data.py --profile-only # Timeframe tertentu python3 scripts/sync_data.py --timeframes 1h,4h,1D # Limit jumlah coin python3 scripts/sync_data.py --limit 10 # Hanya coin yang belum pernah sync python3 scripts/sync_data.py --scope never_synced # Hanya coin yang stale (>24h) python3 scripts/sync_data.py --scope stale # Coin tertentu python3 scripts/sync_data.py --coin bitcoin,ethereum # Dry-run (lihat apa yang akan disync) python3 scripts/sync_data.py --dry-run # Hanya koin yang tersedia di provider (skip 0 records) python3 scripts/sync_data.py --available-only # Quiet mode (hanya summary) python3 scripts/sync_data.py --quiet Cron examples: # Sync kripto setiap jam 0 * * * * cd /path/to/project && python3 scripts/sync_data.py --quiet >> /var/log/sync_crypto.log 2>&1 # Sync saham IDX setiap 30 menit (jam trading) */30 9-16 * * 1-5 cd /path/to/project && python3 scripts/sync_data.py --mode stock --quiet # Sync semua mode jam 00:00 0 0 * * * cd /path/to/project && python3 scripts/sync_data.py --mode all --quiet # Sync saham US jam 22:00-05:00 WIB (jam trading US) 0 22-23,0-5 * * 1-5 cd /path/to/project && python3 scripts/sync_data.py --mode stock_us --quiet """ from __future__ import annotations import argparse import os import sys import time sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from dotenv import load_dotenv load_dotenv() def main(): parser = argparse.ArgumentParser( description='Sync market data (profiles + OHLCV) ke database', formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument('--mode', default='stock', choices=['crypto', 'stock', 'stock_us', 'all'], help='Asset mode (default: stock)') parser.add_argument('--coin', type=str, default='', help='Coin IDs tertentu, comma-separated (e.g. bitcoin,ethereum)') parser.add_argument('--scope', default='', choices=['', 'never_synced', 'stale'], help='Filter: never_synced, stale (>24h), atau kosong (semua)') parser.add_argument('--limit', type=int, default=0, help='Limit jumlah coin (0=semua)') parser.add_argument('--timeframes', type=str, default='', help='Timeframes, comma-separated (e.g. 1h,4h,1D). Kosong=default setting') parser.add_argument('--ohlcv-only', action='store_true', help='Hanya sync OHLCV (skip profile & icon)') parser.add_argument('--profile-only', action='store_true', help='Hanya sync profile & icon (skip OHLCV)') parser.add_argument('--no-icon', action='store_true', help='Skip icon download') parser.add_argument('--available-only', action='store_true', help='Hanya sync koin yang tersedia di provider aktif (skip yg 0 records)') parser.add_argument('--dry-run', action='store_true', help='Tampilkan apa yang akan disync tanpa eksekusi') parser.add_argument('--quiet', '-q', action='store_true', help='Quiet mode — hanya tampilkan summary') parser.add_argument('--verbose', '-v', action='store_true', help='Verbose logging') args = parser.parse_args() if args.verbose: import logging logging.basicConfig(level=logging.DEBUG) from app import create_app app = create_app() modes = ['crypto', 'stock', 'stock_us'] if args.mode == 'all' else [args.mode] total_ok = 0 total_fail = 0 total_coins = 0 grand_start = time.time() for mode in modes: with app.app_context(): # Switch market DB for this mode (tri-database routing) from app.helpers.market_db import switch_market_schema switch_market_schema(mode) ok, fail, coins = _sync_mode(mode, args) total_ok += ok total_fail += fail total_coins += coins elapsed = time.time() - grand_start mins = int(elapsed // 60) secs = int(elapsed % 60) print(f'\n{"=" * 60}') print(f'TOTAL: {total_coins} koin, {total_ok} berhasil, {total_fail} gagal — {mins}m {secs}s') print(f'{"=" * 60}') sys.exit(1 if total_fail > 0 else 0) def _sync_mode(mode: str, args): """Sync satu mode, return (ok_count, fail_count, coin_count).""" from datetime import datetime, timedelta from app.extensions import db from app.models.coin import Coin from app.models.ohlcv import OHLCVData from app.models.settings import AppSettings from app.services.data_sync.sync_manager import SyncManager from sqlalchemy import func mode_labels = {'crypto': 'Kripto', 'stock': 'Saham IDX', 'stock_us': 'Saham US'} label = mode_labels.get(mode, mode) print(f'\n{"─" * 60}') print(f'📦 Sync {label}') print(f'{"─" * 60}') # ── Resolve coin list ── if args.coin: coin_ids = [c.strip() for c in args.coin.split(',') if c.strip()] else: # Filter by asset_type asset_map = {'crypto': 'crypto', 'stock': 'stock', 'stock_us': 'stock_us'} asset_type = asset_map[mode] query = Coin.query.filter_by(is_active=True, asset_type=asset_type) # Scope filtering if args.scope == 'never_synced': coins_with_data = db.session.query( OHLCVData.coin_id ).distinct().subquery() query = query.filter(~Coin.id.in_(db.session.query(coins_with_data))) elif args.scope == 'stale': cutoff = datetime.utcnow() - timedelta(hours=24) recently_synced = db.session.query( OHLCVData.coin_id ).group_by(OHLCVData.coin_id).having( func.max(OHLCVData.created_at) >= cutoff ).subquery() query = query.filter(~Coin.id.in_(db.session.query(recently_synced))) # Order by rank query = query.order_by( db.case((Coin.market_cap_rank.is_(None), 1), else_=0), Coin.market_cap_rank.asc(), ) if args.limit > 0: query = query.limit(args.limit) coin_ids = [c.id for c in query.all()] # ── Filter: available-only ── if args.available_only and coin_ids: original_count = len(coin_ids) coin_ids = _filter_available(coin_ids, mode) skipped = original_count - len(coin_ids) if skipped > 0: print(f' ℹ --available-only: {len(coin_ids)} tersedia, {skipped} dilewati') if not coin_ids: print(f' ⊘ Tidak ada koin {label} untuk disync.') return 0, 0, 0 # ── Resolve timeframes ── if args.profile_only: timeframes = [] # skip OHLCV elif args.timeframes: valid = {'1m', '15m', '30m', '1h', '4h', '1D', '1W'} timeframes = [tf.strip() for tf in args.timeframes.split(',') if tf.strip() in valid] else: # Use saved setting per mode if mode == 'stock_us': timeframes = AppSettings.get('stock_us_sync_timeframes', ['1h', '4h', '1D']) elif mode == 'stock': timeframes = AppSettings.get('stock_sync_timeframes', ['1h', '4h', '1D']) else: timeframes = AppSettings.get('sync_timeframes', ['1h', '4h', '1D']) if timeframes is None: timeframes = ['1h', '4h', '1D'] include_profile = not args.ohlcv_only include_icon = include_profile and not args.no_icon # ── Summary ── tf_str = ', '.join(timeframes) if timeframes else '(skip)' print(f' Koin : {len(coin_ids)}') print(f' Profile: {"Ya" if include_profile else "Tidak"}') print(f' Icon : {"Ya" if include_icon else "Tidak"}') print(f' OHLCV : {tf_str}') if args.scope: print(f' Scope : {args.scope}') print() if args.dry_run: for i, cid in enumerate(coin_ids, 1): coin = db.session.get(Coin, cid) name = coin.name if coin else cid symbol = coin.symbol.upper() if coin and coin.symbol else '?' print(f' [{i:3d}] {symbol:<10} {name}') print(f'\n (dry-run: {len(coin_ids)} koin akan disync)') return 0, 0, len(coin_ids) # ── Execute sync ── manager = SyncManager() queue = manager.build_batch_queue( coin_ids, include_profile=include_profile, include_icon=include_icon, timeframes=timeframes if timeframes else [], ) ok_count = 0 fail_count = 0 current_coin = '' start_time = time.time() for event in manager.execute_sync_queue(queue): evt_type = event.get('type') if evt_type == 'start': if not args.quiet: print(f' ▶ {event["msg"]}') elif evt_type == 'step': key = event.get('key', '') status = event.get('status', '') pct = event.get('pct', 0) msg = event.get('msg', '') # Track coin boundary for concise output coin_id = key.split(':')[0] if ':' in key else '' if coin_id and coin_id != current_coin: current_coin = coin_id if status == 'done': ok_count += 1 if not args.quiet: print(f' ✓ [{pct:3d}%] {msg}') elif status == 'failed': fail_count += 1 if not args.quiet: print(f' ✗ [{pct:3d}%] {msg}') else: # Even in quiet mode, show failures print(f' ✗ {msg}') elif status == 'running' and not args.quiet and args.verbose: print(f' … [{pct:3d}%] {msg}') elif evt_type == 'done': elapsed = time.time() - start_time mins = int(elapsed // 60) secs = int(elapsed % 60) print(f'\n ✅ {event["msg"]} — {mins}m {secs}s') if event.get('failed_items'): print(f' ❌ Gagal:') for fi in event['failed_items'][:10]: print(f' • {fi["label"]}: {fi["error"]}') if len(event['failed_items']) > 10: print(f' ... dan {len(event["failed_items"]) - 10} lainnya') return ok_count, fail_count, len(coin_ids) def _filter_available(coin_ids: list[str], mode: str) -> list[str]: """Filter coin_ids to only those available at the OHLCV provider. Uses local/cached checks — only 1 API call total (pair/coin listing). - Crypto → checks Indodax pairs (real-time tradeable, no rate limit) - Stocks → assumed all available on Yahoo Finance """ if mode in ('stock', 'stock_us'): return coin_ids # Yahoo supports all stocks # Crypto: filter against Indodax pairs (fast, no rate limit for OHLCV) from app.services.data_sync.router import DataRouter router = DataRouter() indodax = router.get_provider('indodax') if not indodax: return coin_ids indodax._get_pairs() # load pairs cache (1 API call) return [cid for cid in coin_ids if indodax._find_pair_by_coin_id(cid)] if __name__ == '__main__': main()