# airflow/plugins/sec_api/client.py
class SECAPIClient:
    BASE_URL = "https://data.sec.gov"
    
    def get_recent_form4_filings(self, cik: str, limit: int = 10):
        """특정 회사의 최근 Form 4 목록"""
        url = f"{self.BASE_URL}/submissions/CIK{cik.zfill(10)}.json"
        
        headers = {
            'User-Agent': 'Mozilla/5.0 ...',  # 필수!
            'Accept': 'application/json'
        }
        
        response = requests.get(url, headers=headers)
        data = response.json()
        
        # Form 4만 필터링
        filings = []
        for i, form_type in enumerate(data['filings']['recent']['form']):
            if form_type == '4':
                filings.append({
                    'accessionNumber': data['filings']['recent']['accessionNumber'][i],
                    'filingDate': data['filings']['recent']['filingDate'][i],
                    'primaryDocument': data['filings']['recent']['primaryDocument'][i]
                })
        
        return filings[:limit]

# airflow/plugins/sec_api/parser.py
class Form4XMLParser:
    @staticmethod
    def _parse_transactions(root) -> List[Dict]:
        """거래 내역 파싱"""
        transactions = []
        
        # nonDerivativeTransaction 요소 찾기
        for trans_elem in root.iter():
            if trans_elem.tag.lower() != 'nonderivativetransaction':
                continue
            
            trans_dict = {}
            
            # 직접 하위 요소 순회
            for child in trans_elem:
                child_tag = child.tag.lower()
                
                # 거래일
                if 'transactiondate' in child_tag:
                    value_elem = child.find('value')
                    if value_elem is not None:
                        trans_dict['transaction_date'] = value_elem.text.strip()
                
                # 거래 코드
                elif 'transactioncoding' in child_tag:
                    code_elem = child.find('transactionCode')
                    if code_elem is not None:
                        trans_dict['transaction_code'] = code_elem.text.strip()
                
                # 주식 수
                elif 'transactionamounts' in child_tag:
                    shares_elem = child.find('.//transactionShares/value')
                    if shares_elem is not None:
                        trans_dict['shares'] = int(float(shares_elem.text))
                    
                    # 가격 (있으면)
                    price_elem = child.find('.//transactionPricePerShare/value')
                    if price_elem is not None:
                        trans_dict['price_per_share'] = float(price_elem.text)
                    else:
                        trans_dict['price_per_share'] = 0.0
            
            if trans_dict.get('shares', 0) > 0:
                transactions.append(trans_dict)
        
        return transactions

# airflow/plugins/price_fetcher.py
import yfinance as yf
from datetime import datetime, timedelta

class StockPriceFetcher:
    @staticmethod
    def get_closing_price(ticker: str, date: str) -> Optional[float]:
        """특정 날짜의 종가 조회"""
        try:
            target_date = datetime.strptime(date, '%Y-%m-%d')
            
            # 주말/휴일 대비 앞뒤 3일 조회
            start_date = target_date - timedelta(days=3)
            end_date = target_date + timedelta(days=3)
            
            # Yahoo Finance에서 데이터 가져오기
            stock = yf.Ticker(ticker)
            hist = stock.history(start=start_date, end=end_date)
            
            if hist.empty:
                return None
            
            # 정확한 날짜 또는 가장 가까운 날짜
            for idx in hist.index:
                if idx.strftime('%Y-%m-%d') == date:
                    return float(hist.loc[idx, 'Close'])
            
            # 가장 가까운 날짜
            closest_idx = min(hist.index, key=lambda x: abs(x - target_date))
            return float(hist.loc[closest_idx, 'Close'])
            
        except Exception as e:
            logger.error(f"Error fetching price: {e}")
            return None

# local_api_crawler.py
price_fetcher = StockPriceFetcher()

for trans in transactions:
    # 가격이 0이고 BUY/SELL인 경우만 API 호출
    if trans.get('price_per_share', 0) == 0 and \
       trans.get('transaction_type') in ['BUY', 'SELL']:
        
        price = price_fetcher.get_closing_price_with_retry(ticker, trans_date)
        
        if price:
            trans['price_per_share'] = price
            trans['transaction_value'] = trans['shares'] * price
            logger.info(f"✅ Updated: ${price:.2f}")

# airflow/dags/daily_insider_ranking.py

def calculate_top_buys(**context):
    """금액 기준 TOP 10 매수"""
    sql = """
        SELECT 
            ticker,
            company_name,
            SUM(shares) as total_shares,
            ROUND(AVG(price_per_share), 2) as avg_price,
            SUM(transaction_value) as total_buy_value,  -- 금액 기준!
            COUNT(*) as buy_count,
            COUNT(DISTINCT insider_name) as insider_count
        FROM insider_trades
        WHERE DATE(created_at_kst) = CURDATE()
          AND transaction_type IN ('BUY', 'OPTION')
          AND transaction_value > 0  -- 가격 있는 것만
        GROUP BY ticker, company_name
        ORDER BY total_buy_value DESC  -- 금액 순 정렬
        LIMIT 10
    """
    
    cursor.execute(sql)
    return cursor.fetchall()

def send_slack_notification(**context):
    """Slack으로 일일 랭킹 전송"""
    
    message = {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": "📊 Daily Insider Trading Report"
                }
            }
        ]
    }
    
    # 매수 랭킹
    for i, row in enumerate(top_buys, 1):
        # 이모지 선택
        if row['total_buy_value'] >= 10000000:
            emoji = "🔥"
        elif row['total_buy_value'] >= 1000000:
            emoji = "⭐"
        else:
            emoji = "💎"
        
        message["blocks"].append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"{i}. {emoji} *{row['ticker']}* - {row['company_name']}\n"
                        f"   💰 *${row['total_buy_value']:,.0f}* "
                        f"({row['total_shares']:,} shares @ ${row['avg_price']})\n"
                        f"   📊 {row['buy_count']} txn(s) | 👥 {row['insider_count']} insider(s)"
            }
        })
    
    # Slack 전송
    requests.post(webhook_url, json=message)

time.sleep(0.15)  # 6.67 req/sec로 제한