"""Merge IDX stock data from official Excel listing into the database. Reads 'Daftar Saham - 20260217.xlsx' and: 1. Updates existing stocks with: listing_date, papan (sub_sector) 2. Creates new stocks that exist in Excel but not in DB 3. Adds Yahoo Finance source mappings for new stocks Usage: python3 scripts/merge_idx_excel.py python3 scripts/merge_idx_excel.py --dry-run """ from __future__ import annotations import sys import os import re from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from dotenv import load_dotenv load_dotenv() # Month name mapping for Indonesian date format MONTH_MAP = { 'jan': 1, 'feb': 2, 'mar': 3, 'apr': 4, 'mei': 5, 'jun': 6, 'jul': 7, 'agu': 8, 'ags': 8, 'aug': 8, 'sep': 9, 'okt': 10, 'nop': 11, 'nov': 11, 'des': 12, } def parse_indo_date(s: str): """Parse Indonesian date like '09 Des 1997' → date(1997, 12, 9).""" if not s or not isinstance(s, str): return None s = s.strip() parts = s.split() if len(parts) != 3: return None try: day = int(parts[0]) month = MONTH_MAP.get(parts[1].lower()[:3]) year = int(parts[2]) if not month: return None return datetime(year, month, day).date() except (ValueError, TypeError): return None def parse_shares(s: str) -> int | None: """Parse shares like '1.924.688.333' → 1924688333.""" if not s or not isinstance(s, str): return None try: return int(s.replace('.', '').replace(',', '').strip()) except (ValueError, TypeError): return None def read_excel(): """Read the IDX Excel file and return list of stock dicts.""" import openpyxl xlsx_path = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'Daftar Saham - 20260217.xlsx' ) if not os.path.exists(xlsx_path): print(f'ERROR: File not found: {xlsx_path}') sys.exit(1) wb = openpyxl.load_workbook(xlsx_path, read_only=True, data_only=True) ws = wb.active stocks = [] for i, row in enumerate(ws.iter_rows(min_row=2, values_only=True)): if not row or not row[1]: continue kode = str(row[1]).strip().upper() nama = str(row[2]).strip() if row[2] else kode tanggal = str(row[3]).strip() if row[3] else '' saham = str(row[4]).strip() if row[4] else '' papan = str(row[5]).strip() if row[5] else '' stocks.append({ 'kode': kode, 'nama': nama, 'listing_date': parse_indo_date(tanggal), 'shares': parse_shares(saham), 'papan': papan, }) wb.close() return stocks def merge(dry_run: bool = False): """Merge Excel data into the database.""" from app import create_app from app.extensions import db from app.models.coin import Coin from app.models.coin_source import CoinSourceMapping from app.helpers.coin_id import make_coin_id excel_stocks = read_excel() print(f'Excel: {len(excel_stocks)} stocks loaded') app = create_app() with app.app_context(): from app.helpers.market_db import switch_market_schema switch_market_schema('stock') # Get existing stock IDs existing = {c.id: c for c in Coin.query.filter_by(asset_type='stock').all()} existing_all = {c.id for c in Coin.query.with_entities(Coin.id).all()} existing_yahoo = {m.coin_id for m in CoinSourceMapping.query .filter_by(source='yahoo') .with_entities(CoinSourceMapping.coin_id).all()} updated = 0 created = 0 skipped = 0 mappings_created = 0 for s in excel_stocks: coin_id = make_coin_id('IDX', s['kode']) if coin_id in existing: # Update existing stock coin = existing[coin_id] changed = False # Update listing_date if not set if s['listing_date'] and not coin.listing_date: coin.listing_date = s['listing_date'] changed = True # Update sub_sector (papan pencatatan) if not set or different if s['papan'] and coin.sub_sector != s['papan']: coin.sub_sector = s['papan'] changed = True # Update name if Excel name is better (more complete) excel_name = s['nama'] if excel_name and len(excel_name) > len(coin.name or ''): # Don't overwrite if current name from Yahoo is longer/better pass # Keep Yahoo name (usually more complete with PT prefix) if changed: updated += 1 if dry_run: print(f' [update] {coin_id}: listing={s["listing_date"]}, papan={s["papan"]}') else: skipped += 1 elif coin_id not in existing_all: # New stock — create it coin = Coin( id=coin_id, symbol=s['kode'], name=s['nama'], is_active=True, asset_type='stock', lot_size=100, listing_date=s['listing_date'], sub_sector=s['papan'], ) if not dry_run: db.session.add(coin) created += 1 # Add Yahoo mapping if coin_id not in existing_yahoo: mapping = CoinSourceMapping( coin_id=coin_id, source='yahoo', source_coin_id=f'{s["kode"]}.JK', source_url=f'https://finance.yahoo.com/quote/{s["kode"]}.JK', is_available=True, ) if not dry_run: db.session.add(mapping) mappings_created += 1 if dry_run: print(f' [new] {coin_id}: {s["nama"]} (listing={s["listing_date"]}, papan={s["papan"]})') else: skipped += 1 if not dry_run: db.session.commit() print(f'\nResults {"(DRY RUN)" if dry_run else ""}:') print(f' Updated: {updated}') print(f' Created: {created}') print(f' Skipped: {skipped} (already up-to-date or non-stock collision)') print(f' Yahoo mappings: {mappings_created}') if not dry_run: total = Coin.query.filter_by(asset_type='stock').count() with_listing = Coin.query.filter( Coin.asset_type == 'stock', Coin.listing_date.isnot(None) ).count() with_papan = Coin.query.filter( Coin.asset_type == 'stock', Coin.sub_sector.isnot(None) ).count() print(f'\n Total stocks: {total}') print(f' With listing_date: {with_listing}') print(f' With papan: {with_papan}') if __name__ == '__main__': dry_run = '--dry-run' in sys.argv merge(dry_run=dry_run)