1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167 | """yfinance용 AWS API Gateway IP rotation 헬퍼.
Yahoo Finance가 IP 기반 레이트리밋을 걸기 시작하면 이 모듈로 전환하세요.
AWS API Gateway 리전의 IP 풀을 이용해 사실상 무제한 IP 로테이션.
**필수 준비사항**:
pip install requests-ip-rotator
AWS 계정 + credentials (~/.aws/credentials 또는 AWS_ACCESS_KEY_ID/SECRET 환경변수)
**비용**:
- AWS 무료 티어: 첫 12개월, 월 1M 요청 무료
- 이후: $3.50 / 1M 요청 (리전당)
- 데이터 전송료 추가 (작은 응답이라 거의 무시 가능)
**리스크**:
- Yahoo가 AWS ASN 대역 전체를 차단하면 무용지물
- 대규모 스크래퍼들은 residential proxy 선호
사용법:
from backend.app.services.yfinance_proxy import ip_rotating_session
import yfinance as yf
with ip_rotating_session() as session:
data = yf.Ticker("AAPL", session=session).history(period="60d")
news = yf.Ticker("AAPL", session=session).news
tickers = yf.Tickers("AAPL MSFT NVDA", session=session)
prices = tickers.history(period="60d", group_by="ticker")
# 컨텍스트 종료 시 gateway 자동 shutdown
"""
from __future__ import annotations
import logging
from contextlib import contextmanager
from typing import Iterator
import requests
logger = logging.getLogger(__name__)
# yfinance 가 사용하는 주요 Yahoo Finance 엔드포인트
_YAHOO_ENDPOINTS = (
"https://query1.finance.yahoo.com",
"https://query2.finance.yahoo.com",
"https://fc.yahoo.com",
)
# 기본 AWS 리전 (IP 풀 다양성 확보를 위해 미국 3개 + 일본 1개)
DEFAULT_REGIONS = (
"us-east-1",
"us-west-2",
"us-east-2",
"ap-northeast-1", # 도쿄 - 일본 주식 데이터 근접성
)
class IpRotationNotAvailable(RuntimeError):
"""requests-ip-rotator 미설치 또는 AWS 설정 누락."""
def _import_rotator():
try:
from requests_ip_rotator import ApiGateway # type: ignore
return ApiGateway
except ImportError as exc:
raise IpRotationNotAvailable(
"requests-ip-rotator not installed. Run: pip install requests-ip-rotator"
) from exc
def create_rotating_session(
endpoints: tuple[str, ...] = _YAHOO_ENDPOINTS,
regions: tuple[str, ...] = DEFAULT_REGIONS,
access_key_id: str | None = None,
access_key_secret: str | None = None,
persistent: bool = True,
) -> tuple[requests.Session, list]:
"""IP rotation 세션 생성.
Args:
endpoints: mount 할 타겟 도메인들 (Yahoo Finance 기본값)
regions: AWS 리전 목록 (IP 풀 다양성)
access_key_id / access_key_secret: 명시적으로 지정하지 않으면 AWS SDK
기본 chain (~/.aws/credentials, env vars) 사용
persistent: True면 gateway를 AWS에 영구 등록 (shutdown 해도 유지) →
다음 실행 시 재사용 가능. AWS CreateRestApi rate limit 우회.
Returns:
(session, gateways) - session 은 mount 완료된 requests.Session,
gateways 는 shutdown_gateways() 에 넘길 ApiGateway 객체 리스트.
"""
ApiGateway = _import_rotator()
session = requests.Session()
gateways = []
for endpoint in endpoints:
kwargs: dict = {"regions": list(regions)}
if access_key_id and access_key_secret:
kwargs["access_key_id"] = access_key_id
kwargs["access_key_secret"] = access_key_secret
gateway = ApiGateway(endpoint, **kwargs)
try:
# require_manual_deletion=True: shutdown 해도 AWS에 남아있음 → 재사용 가능
gateway.start(force=False, require_manual_deletion=persistent)
except Exception as exc: # noqa: BLE001
logger.error("Failed to start gateway for %s: %s", endpoint, exc)
# 이미 시작된 gateway 정리 (persistent 상관없이 session 단위 shutdown)
shutdown_gateways(gateways, persistent=persistent)
raise IpRotationNotAvailable(f"Gateway start failed: {exc}") from exc
session.mount(endpoint, gateway)
gateways.append(gateway)
logger.info("IP rotation gateway started: %s (regions=%s, persistent=%s)", endpoint, regions, persistent)
return session, gateways
def shutdown_gateways(gateways: list, persistent: bool = True) -> None:
"""Gateway 종료. persistent=True면 AWS 리소스 유지 (다음 실행 때 재사용)."""
if persistent:
# AWS에서 삭제하지 않음 — 다음 create_rotating_session에서 재사용됨
logger.info("Gateways kept alive (persistent=True) for reuse")
return
for gateway in gateways:
try:
gateway.shutdown()
except Exception as exc: # noqa: BLE001
logger.warning("Gateway shutdown failed: %s", exc)
@contextmanager
def ip_rotating_session(
endpoints: tuple[str, ...] = _YAHOO_ENDPOINTS,
regions: tuple[str, ...] = DEFAULT_REGIONS,
persistent: bool = True,
) -> Iterator[requests.Session]:
"""컨텍스트 매니저: 세션 사용 후 자동 정리.
persistent=True (기본): AWS gateway를 영구 유지해서 다음 실행 시 재사용.
사용 예:
with ip_rotating_session() as session:
ticker = yf.Ticker("AAPL", session=session)
print(ticker.info)
"""
session, gateways = create_rotating_session(
endpoints=endpoints, regions=regions, persistent=persistent
)
try:
yield session
finally:
shutdown_gateways(gateways, persistent=persistent)
def is_available() -> bool:
"""IP rotation 사용 가능 여부 (패키지 설치 + AWS creds). 실제 시작은 하지 않음."""
try:
_import_rotator()
except IpRotationNotAvailable:
return False
# AWS creds 존재 여부 체크
try:
import boto3 # type: ignore
sts = boto3.client("sts")
sts.get_caller_identity()
return True
except Exception as exc: # noqa: BLE001
logger.debug("AWS credentials unavailable: %s", exc)
return False
|