마이데이터

# 마이데이터 API 연동 예제 (금융 분야)
import requests
from datetime import datetime
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum

class ConsentScope(Enum):
    """마이데이터 동의 범위"""
    ACCOUNT_INFO = "account_info"          # 계좌 정보
    TRANSACTION_HISTORY = "transaction"    # 거래 내역
    CREDIT_SCORE = "credit_score"          # 신용 점수
    INSURANCE = "insurance"                 # 보험 정보
    INVESTMENT = "investment"               # 투자 정보

@dataclass
class MyDataConsent:
    """마이데이터 전송요구 동의"""
    consent_id: str
    user_ci: str                           # 연계정보 (CI)
    scopes: List[ConsentScope]
    target_org: str                        # 정보 제공 기관
    valid_until: datetime
    purpose: str                           # 이용 목적

class MyDataClient:
    """마이데이터 API 클라이언트 (금융보안원 표준 기반)"""

    def __init__(self, client_id: str, client_secret: str,
                 org_code: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.org_code = org_code
        self.base_url = "https://api.mydata.go.kr"
        self.access_token: Optional[str] = None

    def authenticate(self, authorization_code: str) -> str:
        """OAuth 2.0 인증 (사용자 동의 후)"""
        response = requests.post(
            f"{self.base_url}/oauth/2.0/token",
            data={
                "grant_type": "authorization_code",
                "code": authorization_code,
                "client_id": self.client_id,
                "client_secret": self.client_secret
            }
        )
        self.access_token = response.json()["access_token"]
        return self.access_token

    def request_data_transfer(self, consent: MyDataConsent) -> Dict:
        """개인신용정보 전송요구 (신용정보법 제22조의9)"""
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "x-api-tran-id": f"{self.org_code}-{datetime.now().strftime('%Y%m%d%H%M%S')}"
        }

        payload = {
            "user_ci": consent.user_ci,
            "org_code": consent.target_org,
            "scopes": [s.value for s in consent.scopes],
            "valid_period": consent.valid_until.isoformat(),
            "purpose": consent.purpose
        }

        response = requests.post(
            f"{self.base_url}/v1/transfer/request",
            headers=headers,
            json=payload
        )

        return response.json()

    def get_account_list(self, user_ci: str, org_code: str) -> List[Dict]:
        """계좌 목록 조회"""
        headers = {"Authorization": f"Bearer {self.access_token}"}
        response = requests.get(
            f"{self.base_url}/v1/bank/accounts",
            headers=headers,
            params={"org_code": org_code}
        )
        return response.json().get("accounts", [])

    def get_transactions(self, account_num: str,
                          from_date: str, to_date: str) -> List[Dict]:
        """거래 내역 조회"""
        headers = {"Authorization": f"Bearer {self.access_token}"}
        response = requests.get(
            f"{self.base_url}/v1/bank/accounts/{account_num}/transactions",
            headers=headers,
            params={"from_date": from_date, "to_date": to_date}
        )
        return response.json().get("transactions", [])

    def aggregate_financial_data(self, user_ci: str) -> Dict:
        """사용자 금융 데이터 통합 조회 (AI 분석용)"""
        # 여러 기관의 데이터 수집
        aggregated = {
            "user_ci": user_ci,
            "collected_at": datetime.now().isoformat(),
            "accounts": [],
            "transactions": [],
            "total_assets": 0
        }

        # 실제로는 동의받은 모든 기관에서 데이터 수집
        banks = ["BANK001", "BANK002"]  # 동의받은 기관 코드

        for bank in banks:
            accounts = self.get_account_list(user_ci, bank)
            for account in accounts:
                aggregated["accounts"].append(account)
                aggregated["total_assets"] += account.get("balance", 0)

        return aggregated

# AI 개인화 서비스 예시
def recommend_financial_product(user_data: Dict) -> Dict:
    """마이데이터 기반 AI 금융상품 추천"""
    total_assets = user_data.get("total_assets", 0)
    transactions = user_data.get("transactions", [])

    # 간단한 규칙 기반 추천 (실제로는 ML 모델 사용)
    recommendation = {
        "user_ci": user_data["user_ci"],
        "timestamp": datetime.now().isoformat(),
        "products": [],
        "explanation": ""  # AI Act 투명성 요건: 추천 근거 설명
    }

    if total_assets > 50000000:
        recommendation["products"].append({
            "type": "investment",
            "name": "고액자산 ETF 포트폴리오",
            "reason": "총 자산 5천만원 이상 고객 대상"
        })

    recommendation["explanation"] = (
        f"귀하의 총 자산 {total_assets:,}원과 "
        f"최근 거래 패턴을 분석하여 추천되었습니다."
    )

    return recommendation

# 사용 예시
client = MyDataClient("client_id", "secret", "MYDATA001")
# client.authenticate(auth_code)
# data = client.aggregate_financial_data("user_ci_value")
# recommendation = recommend_financial_product(data)