"""SyncManager - orchestrates data sync with UPSERT logic.""" from __future__ import annotations import logging from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from enum import Enum from typing import Optional from flask import current_app from sqlalchemy.dialects.mysql import insert as mysql_insert from app.extensions import db from app.models.asset import Asset, AssetProfile from app.models.asset_source import AssetSourceMapping from app.models.ohlcv import OHLCVData from app.models.ticker import MarketTicker from app.models.settings import AppSettings from app.services.data_sync.router import DataRouter logger = logging.getLogger(__name__) # WIB timezone (UTC+7) WIB = timezone(timedelta(hours=7)) # ─── SyncQueue Types ───────────────────────────────────────────────────────── class SyncItemStatus(Enum): PENDING = 'pending' RUNNING = 'running' DONE = 'done' FAILED = 'failed' RETRYING = 'retrying' @dataclass class SyncItem: key: str # e.g. 'profile', 'icon', 'ohlcv_1h' label: str # Display label asset_id: str item_type: str # 'profile', 'icon', 'ohlcv' timeframe: str = '' status: SyncItemStatus = SyncItemStatus.PENDING attempts: int = 0 max_attempts: int = 2 result: dict = field(default_factory=dict) error: str = '' class SyncQueue: """In-memory sync queue with tracking.""" def __init__(self): self.items: list[SyncItem] = [] @property def total(self): return len(self.items) @property def completed(self): return sum(1 for i in self.items if i.status == SyncItemStatus.DONE) @property def failed(self): return sum(1 for i in self.items if i.status == SyncItemStatus.FAILED) @property def pending(self): return sum(1 for i in self.items if i.status in (SyncItemStatus.PENDING, SyncItemStatus.RETRYING)) @property def all_done(self): return all(i.status in (SyncItemStatus.DONE, SyncItemStatus.FAILED) for i in self.items) def add(self, item: SyncItem): self.items.append(item) def get_pending(self) -> list[SyncItem]: return [i for i in self.items if i.status in (SyncItemStatus.PENDING, SyncItemStatus.RETRYING)] def to_summary(self) -> dict: return { 'total': self.total, 'completed': self.completed, 'failed': self.failed, 'items': [ { 'key': i.key, 'label': i.label, 'asset_id': i.asset_id, 'status': i.status.value, 'attempts': i.attempts, 'error': i.error, 'result': i.result, } for i in self.items ], } # ─── SyncManager ───────────────────────────────────────────────────────────── class SyncManager: """Manages data synchronization from providers to database.""" def __init__(self): self.router = DataRouter() # ── Queue builders ── def build_sync_queue(self, asset_id: str, include_profile: bool = True, include_icon: bool = True, timeframes: list[str] | None = None) -> SyncQueue: """Build a SyncQueue for a single asset. Args: timeframes: Specific OHLCV timeframes to sync. ``None`` uses the saved ``sync_timeframes`` setting. An empty list ``[]`` skips OHLCV entirely. """ queue = SyncQueue() max_attempts = current_app.config.get('SYNC_ITEM_MAX_ATTEMPTS', 2) if include_profile: queue.add(SyncItem( key='profile', label='Profile & Market Data', asset_id=asset_id, item_type='profile', max_attempts=max_attempts, )) if include_icon: queue.add(SyncItem( key='icon', label='Download Icon', asset_id=asset_id, item_type='icon', max_attempts=max_attempts, )) if timeframes is None: timeframes = AppSettings.get('sync_timeframes', ['1h', '4h', '1D']) for tf in timeframes: queue.add(SyncItem( key=f'ohlcv_{tf}', label=f'OHLCV {tf}', asset_id=asset_id, item_type='ohlcv', timeframe=tf, max_attempts=max_attempts, )) return queue def build_batch_queue(self, asset_ids: list[str], include_profile: bool = True, include_icon: bool = True, timeframes: list[str] | None = None) -> SyncQueue: """Build a SyncQueue for multiple assets. Uses bulk query to avoid N+1 and efficiently handles large asset lists. Args: timeframes: Specific OHLCV timeframes to sync. ``None`` uses the saved ``sync_timeframes`` setting. An empty list ``[]`` skips OHLCV entirely. """ queue = SyncQueue() max_attempts = current_app.config.get('SYNC_ITEM_MAX_ATTEMPTS', 2) if timeframes is None: from app.helpers.asset_filter import get_sync_timeframes try: timeframes = get_sync_timeframes() except RuntimeError: timeframes = AppSettings.get('sync_timeframes', ['1h', '4h', '1D']) # Bulk fetch asset names in one query (avoid N+1) coin_names = {} if asset_ids: rows = db.session.query(Asset.id, Asset.name).filter( Asset.id.in_(asset_ids) ).all() coin_names = {r[0]: r[1] for r in rows} # Determine which assets are stocks (need extended data) stock_ids = set() if asset_ids: stock_rows = db.session.query(Asset.id, Asset.asset_type).filter( Asset.id.in_(asset_ids), Asset.asset_type.in_(['stock', 'stock_us']) ).all() stock_ids = {r[0] for r in stock_rows} for asset_id in asset_ids: coin_name = coin_names.get(asset_id, asset_id) if include_profile: queue.add(SyncItem( key=f'{asset_id}:profile', label=f'{coin_name} — Profile', asset_id=asset_id, item_type='profile', max_attempts=max_attempts, )) if include_icon: queue.add(SyncItem( key=f'{asset_id}:icon', label=f'{coin_name} — Icon', asset_id=asset_id, item_type='icon', max_attempts=max_attempts, )) # Extended data step for stocks (fetches all: financials, earnings, # dividends, analyst, ownership, news, options in one step) if asset_id in stock_ids: queue.add(SyncItem( key=f'{asset_id}:extended', label=f'{coin_name} — Extended Data', asset_id=asset_id, item_type='extended', max_attempts=max_attempts, )) for tf in timeframes: queue.add(SyncItem( key=f'{asset_id}:ohlcv_{tf}', label=f'{coin_name} — OHLCV {tf}', asset_id=asset_id, item_type='ohlcv', timeframe=tf, max_attempts=max_attempts, )) return queue # ── Queue executor (generator for SSE) ── def execute_sync_queue(self, queue: SyncQueue, provider_id: str | None = None): """Execute all items in the queue, yielding progress dicts for SSE. Yields: dict events: - {type: 'start', steps: [...], total_steps: N} - {type: 'step', key, status, step, total_steps, attempt, is_retry, pct, msg} - {type: 'done', all_ok, summary, pct: 100} """ # Default provider (can be overridden per-asset below) if provider_id: default_provider = self.router.get_provider(provider_id) if not default_provider: default_provider = self.router.get_active_provider() provider_id = self.router.get_active_provider_id() default_provider_id = provider_id else: default_provider = self.router.get_active_provider() default_provider_id = self.router.get_active_provider_id() total_steps = queue.total # Build asset groups summary for lightweight start event # (don't send 90k individual steps — send grouped asset summary) coin_groups = {} for item in queue.items: if item.asset_id not in coin_groups: coin_name = item.label.split(' \u2014 ')[0] if ' \u2014 ' in item.label else item.asset_id coin_groups[item.asset_id] = {'name': coin_name, 'count': 0} coin_groups[item.asset_id]['count'] += 1 yield { 'type': 'start', 'total_steps': total_steps, 'coin_count': len(coin_groups), 'assets': [ {'asset_id': cid, 'name': info['name'], 'step_count': info['count']} for cid, info in coin_groups.items() ], 'msg': f'Memulai sync ({len(coin_groups)} koin, {total_steps} item)...', } step_num = 0 # Process pending items while queue.pending > 0: for item in queue.get_pending(): item.attempts += 1 is_retry = item.attempts > 1 item.status = SyncItemStatus.RUNNING step_num += 1 pct = int((step_num - 1) / total_steps * 100) if total_steps > 0 else 0 yield { 'type': 'step', 'key': item.key, 'status': 'running', 'step': step_num, 'total_steps': total_steps, 'attempt': item.attempts, 'is_retry': is_retry, 'pct': pct, 'msg': (f'Retry #{item.attempts - 1}: ' if is_retry else '') + f'{item.label}...', } ok = False detail = '' # Resolve provider per-asset (stocks → yahoo, crypto → active) item_provider, item_provider_id = self.router.get_provider_for_coin(item.asset_id) if not item_provider: item_provider, item_provider_id = default_provider, default_provider_id try: if item.item_type == 'profile': ok, detail = self._execute_profile_sync( item.asset_id, item_provider, item_provider_id) elif item.item_type == 'icon': ok, detail = self._execute_icon_sync(item.asset_id) elif item.item_type == 'ohlcv': ok, detail = self._execute_ohlcv_sync( item.asset_id, item.timeframe, item_provider, item_provider_id) elif item.item_type == 'extended': ok, detail = self._execute_extended_sync( item.asset_id, item_provider) except Exception as e: detail = str(e)[:120] logger.error( 'Sync error %s/%s: %s', item.asset_id, item.key, e) if ok: item.status = SyncItemStatus.DONE item.result = {'detail': detail} elif item.attempts < item.max_attempts: item.status = SyncItemStatus.RETRYING item.error = detail else: item.status = SyncItemStatus.FAILED item.error = detail pct = int(step_num / total_steps * 100) if total_steps > 0 else 0 yield { 'type': 'step', 'key': item.key, 'status': item.status.value, 'step': step_num, 'total_steps': total_steps, 'attempt': item.attempts, 'is_retry': is_retry, 'pct': pct, 'msg': f'{item.label}: {detail}', 'detail': detail, } # If there are still retrying items, they'll be picked up next loop # Done — only include failed items in summary (not 90k items) all_ok = queue.failed == 0 failed_items = [ {'key': i.key, 'label': i.label, 'asset_id': i.asset_id, 'error': i.error, 'attempts': i.attempts} for i in queue.items if i.status == SyncItemStatus.FAILED ] yield { 'type': 'done', 'all_ok': all_ok, 'ok_count': queue.completed, 'failed_count': queue.failed, 'total_count': queue.total, 'failed_items': failed_items, 'pct': 100, 'msg': ( f'Sync selesai! {queue.completed}/{queue.total} berhasil.' + (f' {queue.failed} gagal.' if queue.failed else '') ), } # ── Item executors ── def _execute_profile_sync(self, asset_id: str, provider, provider_id: str) -> tuple[bool, str]: """Sync profile for a asset. Returns (ok, detail). For crypto: if active provider (e.g. Indodax) doesn't provide profiles, fallback to CoinGecko which has comprehensive profile data. """ profile_data = provider.get_coin_profile(asset_id) if not profile_data and provider_id != 'coingecko': # Fallback to CoinGecko for profile data cg = self.router.get_provider('coingecko') if cg: profile_data = cg.get_coin_profile(asset_id) if not profile_data: return False, 'Tidak ada data profile' self._upsert_coin(profile_data) self._save_profile(asset_id, profile_data) self._save_tickers(asset_id, profile_data.get('tickers', [])) self._update_source_mappings(asset_id) price = profile_data.get('current_price_idr') rank = profile_data.get('market_cap_rank') parts = [] if rank: parts.append(f'Rank #{rank}') if price: parts.append(f'Rp {price:,.0f}') return True, ', '.join(parts) if parts else 'Profile OK' def _execute_icon_sync(self, asset_id: str) -> tuple[bool, str]: """Download icon for a asset. Returns (ok, detail).""" from app.services.icon_downloader import ( download_icon, download_us_stock_icon, get_stock_remote_icon_url, has_local_icon, ) fresh_coin = db.session.get(Asset, asset_id) if not fresh_coin: return False, 'Asset tidak ditemukan' asset_type = fresh_coin.asset_type or 'crypto' symbol = fresh_coin.symbol or '' # For stocks: use dedicated remote URL builder (not local icon_thumb_url) if asset_type == 'stock_us': # US stocks have dual-source fallback (FMP → nvstly) ok = download_us_stock_icon(asset_id, symbol, force=False) if ok: self._update_icon_thumb_url(fresh_coin, asset_id, asset_type) return ok, 'Icon OK' if ok else 'Download gagal' if asset_type == 'stock': # IDX stocks: skip if local icon already exists if has_local_icon(asset_id, asset_type): return True, 'Icon sudah ada' remote_url = get_stock_remote_icon_url(symbol or asset_id) ok = download_icon(asset_id, remote_url, force=True, asset_type=asset_type) if ok: self._update_icon_thumb_url(fresh_coin, asset_id, asset_type) return ok, 'Icon didownload' if ok else 'Download gagal' # Crypto: use icon_thumb_url or image_url from CoinGecko profile remote_url = fresh_coin.icon_thumb_url or fresh_coin.image_url if not remote_url: return False, 'Tidak ada URL icon' # If icon_thumb_url is already a local path, check if file exists if remote_url.startswith('/static/'): if has_local_icon(asset_id, asset_type): return True, 'Icon sudah ada' # Local path but file missing — try image_url instead remote_url = fresh_coin.image_url if not remote_url: return False, 'Tidak ada URL icon remote' ok = download_icon(asset_id, remote_url, force=True, asset_type=asset_type) if ok: self._update_icon_thumb_url(fresh_coin, asset_id, asset_type) return ok, 'Icon didownload' if ok else 'Download gagal' def _update_icon_thumb_url(self, asset, asset_id: str, asset_type: str): """Update asset's icon_thumb_url to local path after successful download.""" from app.services.icon_downloader import get_local_icon_url local_url = get_local_icon_url(asset_id, asset_type) if local_url and asset.icon_thumb_url != local_url: asset.icon_thumb_url = local_url db.session.commit() def _execute_ohlcv_sync(self, asset_id: str, timeframe: str, provider, provider_id: str) -> tuple[bool, str]: """Sync OHLCV for a specific timeframe. Returns (ok, detail).""" now = datetime.now() count = self._sync_ohlcv(asset_id, timeframe, provider_id, provider, now) return True, f'{count:,} records' def _execute_extended_sync(self, asset_id: str, provider) -> tuple[bool, str]: """Sync all extended data for a stock in one step. Fetches: financials, earnings, dividends, analyst, ownership, news, options. Returns (ok, detail). """ import time as _time total_saved = 0 # 1. Financial Statements (6 types) try: fin_data = provider.get_financial_statements(asset_id) if fin_data: self._save_extended_batch(asset_id, fin_data) total_saved += sum(1 for v in fin_data.values() if v) except Exception as e: logger.warning(f'Extended sync financials failed for {asset_id}: {e}') _time.sleep(0.3) # 2. Earnings try: earn_data = provider.get_earnings_data(asset_id) if earn_data: self._save_extended_batch(asset_id, earn_data) total_saved += sum(1 for v in earn_data.values() if v) except Exception as e: logger.warning(f'Extended sync earnings failed for {asset_id}: {e}') _time.sleep(0.3) # 3. Dividends & Splits try: div_data = provider.get_dividends_splits(asset_id) if div_data: self._save_extended_batch(asset_id, div_data) total_saved += sum(1 for v in div_data.values() if v) except Exception as e: logger.warning(f'Extended sync dividends failed for {asset_id}: {e}') _time.sleep(0.3) # 4. Analyst Data try: ana_data = provider.get_analyst_data(asset_id) if ana_data: self._save_extended_batch(asset_id, ana_data) total_saved += sum(1 for v in ana_data.values() if v) except Exception as e: logger.warning(f'Extended sync analyst failed for {asset_id}: {e}') _time.sleep(0.3) # 5. Ownership & Insider try: own_data = provider.get_ownership_data(asset_id) if own_data: self._save_extended_batch(asset_id, own_data) total_saved += sum(1 for v in own_data.values() if v) except Exception as e: logger.warning(f'Extended sync ownership failed for {asset_id}: {e}') _time.sleep(0.3) # 6. News try: news_data = provider.get_news(asset_id) if news_data: self._save_extended_data(asset_id, 'news', news_data) total_saved += 1 except Exception as e: logger.warning(f'Extended sync news failed for {asset_id}: {e}') _time.sleep(0.3) # 7. Options Chain try: opt_data = provider.get_options_data(asset_id) if opt_data: self._save_extended_batch(asset_id, opt_data) total_saved += sum(1 for v in opt_data.values() if v) except Exception as e: logger.warning(f'Extended sync options failed for {asset_id}: {e}') _time.sleep(0.3) # 8. Analyst Estimates (growth, revenue, earnings, EPS) try: est_data = provider.get_estimate_data(asset_id) if est_data: self._save_extended_batch(asset_id, est_data) total_saved += sum(1 for v in est_data.values() if v) except Exception as e: logger.warning(f'Extended sync estimates failed for {asset_id}: {e}') _time.sleep(0.3) # 9. ESG / Sustainability try: sus_data = provider.get_sustainability(asset_id) if sus_data: self._save_extended_batch(asset_id, sus_data) total_saved += sum(1 for v in sus_data.values() if v) except Exception as e: logger.warning(f'Extended sync sustainability failed for {asset_id}: {e}') _time.sleep(0.3) # 10. SEC Filings try: sec_data = provider.get_sec_filings(asset_id) if sec_data: self._save_extended_batch(asset_id, sec_data) total_saved += sum(1 for v in sec_data.values() if v) except Exception as e: logger.warning(f'Extended sync sec_filings failed for {asset_id}: {e}') return total_saved > 0, f'{total_saved} data types saved' # ── Legacy methods (kept for backward compatibility) ── def sync_coin(self, asset_id: str, provider_id: str = None) -> dict: """Sync a single asset: profile + OHLCV for all configured timeframes. Args: asset_id: Asset ID provider_id: Force specific provider (None = auto-detect from asset_type) Returns: Summary dict with sync results """ if provider_id: provider = self.router.get_provider(provider_id) else: # Auto-detect provider from asset's asset_type provider, provider_id = self.router.get_provider_for_coin(asset_id) if not provider: return {'error': f'Provider {provider_id} not found'} results = {'asset_id': asset_id, 'provider': provider_id, 'ohlcv': {}, 'profile': False} # Sync profile profile_data = provider.get_coin_profile(asset_id) if profile_data: self._upsert_coin(profile_data) self._save_profile(asset_id, profile_data) self._save_tickers(asset_id, profile_data.get('tickers', [])) results['profile'] = True # Auto-populate source mappings self._update_source_mappings(asset_id) # Sync OHLCV for configured timeframes (stock-aware) asset = db.session.get(Asset, asset_id) if asset and asset.asset_type in ('stock', 'stock_us'): timeframes = AppSettings.get('stock_sync_timeframes', ['1h', '4h', '1D']) else: timeframes = AppSettings.get('sync_timeframes', ['1h', '4h', '1D']) now = datetime.now() for tf in timeframes: count = self._sync_ohlcv(asset_id, tf, provider_id, provider, now) results['ohlcv'][tf] = count return results def sync_ohlcv_only(self, asset_id: str, timeframe: str, provider_id: str = None) -> int: """Sync OHLCV data for a specific asset and timeframe. Returns: Number of records upserted """ if provider_id: provider = self.router.get_provider(provider_id) else: # Auto-detect provider from asset's asset_type provider, provider_id = self.router.get_provider_for_coin(asset_id) if not provider: return 0 return self._sync_ohlcv(asset_id, timeframe, provider_id, provider, datetime.now()) def _sync_ohlcv(self, asset_id: str, timeframe: str, provider_id: str, provider, now: datetime) -> int: """Internal: fetch and UPSERT OHLCV data.""" # Determine date range based on timeframe tf_config = current_app.config.get('TIMEFRAMES', {}).get(timeframe, {}) minutes = tf_config.get('minutes', 60) if minutes <= 30: start = now - timedelta(days=2) elif minutes <= 240: start = now - timedelta(days=90) else: start = now - timedelta(days=365) # Fetch from provider raw_data = provider.get_ohlcv(asset_id, timeframe, start, now) if not raw_data: return 0 # UPSERT into database return self._upsert_ohlcv_batch(asset_id, timeframe, provider_id, raw_data) def _upsert_ohlcv_batch(self, asset_id: str, timeframe: str, source: str, records: list[dict]) -> int: """Batch UPSERT OHLCV records using INSERT ... ON DUPLICATE KEY UPDATE.""" import math if not records: return 0 rows = [] skipped = 0 for r in records: ts = r['timestamp'] o, h, l, c = r['open'], r['high'], r['low'], r['close'] v = r.get('volume', 0) # Safety net: skip NaN/Inf values that would crash MySQL vals = [o, h, l, c, v] if any(v2 is None or (isinstance(v2, float) and (math.isnan(v2) or math.isinf(v2))) for v2 in vals): skipped += 1 continue # Convert timestamp to WIB datetime dt_utc = datetime.fromtimestamp(ts, tz=timezone.utc) dt_wib = dt_utc.astimezone(WIB).replace(tzinfo=None) rows.append({ 'asset_id': asset_id, 'timeframe': timeframe, 'source': source, 'timestamp': ts, 'datetime_wib': dt_wib, 'open': o, 'high': h, 'low': l, 'close': c, 'volume': v, }) if skipped: logger.warning(f'[sync] {asset_id}/{timeframe}: skipped {skipped} rows with NaN/Inf values') # Batch insert with ON DUPLICATE KEY UPDATE batch_size = 500 total_upserted = 0 for i in range(0, len(rows), batch_size): batch = rows[i:i + batch_size] stmt = mysql_insert(OHLCVData).values(batch) stmt = stmt.on_duplicate_key_update( open=stmt.inserted.open, high=stmt.inserted.high, low=stmt.inserted.low, close=stmt.inserted.close, volume=stmt.inserted.volume, ) db.session.execute(stmt) total_upserted += len(batch) db.session.commit() return total_upserted def _upsert_coin(self, profile: dict): """Insert or update asset master data.""" asset = Asset.query.get(profile['id']) if asset: asset.symbol = profile.get('symbol', asset.symbol) asset.name = profile.get('name', asset.name) asset.description = profile.get('description', asset.description) asset.categories = profile.get('categories', asset.categories) asset.market_cap_rank = profile.get('market_cap_rank', asset.market_cap_rank) asset.image_url = profile.get('image_url', asset.image_url) asset.icon_thumb_url = profile.get('icon_thumb_url', asset.icon_thumb_url) asset.website = profile.get('website', asset.website) asset.blockchain = profile.get('blockchain', asset.blockchain) asset.contract_address = profile.get('contract_address', asset.contract_address) asset.coingecko_score = profile.get('coingecko_score', asset.coingecko_score) # Stock-specific fields if profile.get('asset_type'): asset.asset_type = profile['asset_type'] if profile.get('sector'): asset.sector = profile['sector'] if profile.get('sub_sector'): asset.sub_sector = profile['sub_sector'] if profile.get('lot_size') is not None: asset.lot_size = profile['lot_size'] if profile.get('listing_date') and isinstance(profile['listing_date'], str): try: asset.listing_date = datetime.strptime(profile['listing_date'], '%Y-%m-%d').date() except (ValueError, TypeError): pass if profile.get('genesis_date'): try: asset.genesis_date = datetime.strptime(profile['genesis_date'], '%Y-%m-%d').date() except (ValueError, TypeError): pass else: genesis = None if profile.get('genesis_date'): try: genesis = datetime.strptime(profile['genesis_date'], '%Y-%m-%d').date() except (ValueError, TypeError): pass asset = Asset( id=profile['id'], symbol=profile.get('symbol', ''), name=profile.get('name', ''), description=profile.get('description'), categories=profile.get('categories'), genesis_date=genesis, market_cap_rank=profile.get('market_cap_rank'), image_url=profile.get('image_url'), icon_thumb_url=profile.get('icon_thumb_url'), website=profile.get('website'), blockchain=profile.get('blockchain'), contract_address=profile.get('contract_address'), coingecko_score=profile.get('coingecko_score'), asset_type=profile.get('asset_type', 'crypto'), sector=profile.get('sector'), sub_sector=profile.get('sub_sector'), lot_size=profile.get('lot_size', 1), ) db.session.add(asset) db.session.commit() @staticmethod def _clean_nan(val): """Convert NaN/Inf floats to None for MySQL safety.""" import math if val is None: return None if isinstance(val, float) and (math.isnan(val) or math.isinf(val)): return None return val def _save_profile(self, asset_id: str, profile: dict): """Save a asset profile snapshot.""" _c = self._clean_nan cp = AssetProfile( asset_id=asset_id, current_price_idr=_c(profile.get('current_price_idr')), current_price_usd=_c(profile.get('current_price_usd')), market_cap_idr=_c(profile.get('market_cap_idr')), total_volume_idr=_c(profile.get('total_volume_idr')), circulating_supply=_c(profile.get('circulating_supply')), total_supply=_c(profile.get('total_supply')), max_supply=_c(profile.get('max_supply')), price_change_1h=_c(profile.get('price_change_1h')), price_change_24h=_c(profile.get('price_change_24h')), price_change_7d=_c(profile.get('price_change_7d')), price_change_30d=_c(profile.get('price_change_30d')), ath_idr=_c(profile.get('ath_idr')), atl_idr=_c(profile.get('atl_idr')), twitter_followers=_c(profile.get('twitter_followers')), reddit_subscribers=_c(profile.get('reddit_subscribers')), telegram_users=_c(profile.get('telegram_users')), github_stars=_c(profile.get('github_stars')), sentiment_votes_up=_c(profile.get('sentiment_votes_up')), profile_json=profile, ) db.session.add(cp) db.session.commit() def _save_tickers(self, asset_id: str, tickers: list[dict]): """Save market tickers for a asset (replace old ones).""" # Delete old tickers for this asset MarketTicker.query.filter_by(asset_id=asset_id).delete() _c = self._clean_nan for t in tickers: if t.get('is_stale'): continue ticker = MarketTicker( asset_id=asset_id, market_name=t.get('market'), pair=t.get('pair'), price_idr=_c(t.get('price_idr')), volume_idr=_c(t.get('volume_idr')), spread_pct=_c(t.get('spread_pct')), trust_score=t.get('trust_score'), trade_url=t.get('trade_url'), ) db.session.add(ticker) db.session.commit() def _update_source_mappings(self, asset_id: str): """Check availability across relevant providers and save/update mappings. Only checks providers appropriate for the asset's asset type: - stock/stock_us → Yahoo only - crypto → all providers (CoinGecko, Indodax, Yahoo) """ from app.helpers.asset_id import get_raw_id, parse_asset_id, NAMESPACES raw = get_raw_id(asset_id) now = datetime.now() # Filter providers by asset type to avoid unnecessary API calls prefix, _ = parse_asset_id(asset_id) asset_type = NAMESPACES.get(prefix, 'crypto') if asset_type in ('stock', 'stock_us'): providers_to_check = {'yahoo': self.router.get_provider('yahoo')} else: # Crypto: only check CoinGecko + Indodax (not Yahoo) providers_to_check = { k: v for k, v in self.router.get_all_providers().items() if k != 'yahoo' } for provider in providers_to_check.values(): pid = provider.provider_id existing = AssetSourceMapping.query.filter_by( asset_id=asset_id, source=pid ).first() available = provider.is_available(asset_id) if existing: existing.is_available = available existing.last_checked = now elif available: # Build source-specific info (use raw ID for provider) source_asset_id = raw source_pair = None source_url = None if pid == 'coingecko': source_asset_id = raw source_url = f'https://www.coingecko.com/id/assets/{raw}' elif pid == 'indodax': pair_info = provider._find_pair_by_asset_id(asset_id) if pair_info: source_asset_id = pair_info.get('ticker_id') or pair_info.get('id', raw) source_pair = f"{pair_info['traded_currency']}/IDR" symbol_lower = pair_info['traded_currency'].lower() source_url = f'https://indodax.com/market/{symbol_lower}idr' elif pid == 'yahoo': source_asset_id = f'{raw.upper()}.JK' source_pair = f'{raw.upper()}/IDR' source_url = f'https://finance.yahoo.com/quote/{raw.upper()}.JK' mapping = AssetSourceMapping( asset_id=asset_id, source=pid, source_asset_id=str(source_asset_id), source_pair=source_pair, source_url=source_url, is_available=True, last_checked=now, ) db.session.add(mapping) db.session.commit() # ── Extended data persistence ── def _save_extended_data(self, asset_id: str, data_type: str, data_json): """UPSERT a single extended data record (replaces existing for same asset+type).""" from app.models.asset_extended_data import AssetExtendedData if not data_json: return stmt = mysql_insert(AssetExtendedData).values( asset_id=asset_id, data_type=data_type, data_json=data_json, fetched_at=datetime.utcnow(), ) stmt = stmt.on_duplicate_key_update( data_json=stmt.inserted.data_json, fetched_at=stmt.inserted.fetched_at, ) db.session.execute(stmt) db.session.commit() def _save_extended_batch(self, asset_id: str, data_dict: dict): """Save multiple extended data types at once from a fetched dict.""" saved = 0 for data_type, data_json in data_dict.items(): if data_json: self._save_extended_data(asset_id, data_type, data_json) saved += 1 if saved: logger.info(f'Saved {saved} extended data records for {asset_id}') def ensure_coin_exists(self, asset_id: str, symbol: str = '', name: str = '', asset_type: str = 'crypto') -> Asset: """Ensure a asset exists in the database, create if not.""" from app.helpers.asset_id import get_raw_id asset = Asset.query.get(asset_id) if not asset: raw = get_raw_id(asset_id) lot_size = 100 if asset_type == 'stock' else 1 # stock_us uses lot_size=1 asset = Asset(id=asset_id, symbol=symbol or raw.upper(), name=name or raw.title(), asset_type=asset_type, lot_size=lot_size) db.session.add(asset) db.session.commit() return asset