265 lines
7.9 KiB
Python
265 lines
7.9 KiB
Python
"""
|
|
코드 리뷰 개선사항 검증 스크립트
|
|
|
|
실행 방법:
|
|
python verify_improvements.py
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# 프로젝트 루트를 sys.path에 추가
|
|
project_root = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
|
|
def test_rate_limiter():
|
|
"""Rate Limiter 동작 테스트"""
|
|
print("\n" + "=" * 70)
|
|
print("TEST 1: Rate Limiter 동작 확인")
|
|
print("=" * 70)
|
|
|
|
from src.common import RateLimiter
|
|
|
|
limiter = RateLimiter(max_calls=3, period=1.0)
|
|
|
|
# 3회 연속 호출 (즉시 통과)
|
|
start = time.time()
|
|
for i in range(3):
|
|
limiter.acquire()
|
|
print(f" 호출 {i + 1}: 즉시 통과 (경과: {time.time() - start:.2f}초)")
|
|
|
|
# 4번째 호출 (대기 필요)
|
|
print(" 호출 4: 대기 중...")
|
|
limiter.acquire()
|
|
elapsed = time.time() - start
|
|
print(f" 호출 4: 통과 (경과: {elapsed:.2f}초)")
|
|
|
|
if elapsed >= 1.0:
|
|
print("✅ PASS: Rate Limiter가 정상적으로 호출을 제한했습니다")
|
|
return True
|
|
else:
|
|
print("❌ FAIL: Rate Limiter가 호출을 제한하지 않았습니다")
|
|
return False
|
|
|
|
|
|
def test_config_validation():
|
|
"""설정 파일 검증 테스트"""
|
|
print("\n" + "=" * 70)
|
|
print("TEST 2: 설정 파일 검증")
|
|
print("=" * 70)
|
|
|
|
from src.config import validate_config
|
|
|
|
# 정상 설정
|
|
valid_config = {
|
|
"buy_check_interval_minutes": 240,
|
|
"stop_loss_check_interval_minutes": 60,
|
|
"profit_taking_check_interval_minutes": 240,
|
|
"dry_run": True,
|
|
"auto_trade": {"enabled": False},
|
|
}
|
|
|
|
is_valid, msg = validate_config(valid_config)
|
|
print(f" 정상 설정 검증: {'✅ PASS' if is_valid else '❌ FAIL'}")
|
|
if not is_valid:
|
|
print(f" 오류: {msg}")
|
|
|
|
# 잘못된 설정 1: 필수 항목 누락
|
|
invalid_config1 = {
|
|
"buy_check_interval_minutes": 240,
|
|
"dry_run": True,
|
|
# stop_loss_check_interval_minutes 누락
|
|
}
|
|
|
|
is_valid, msg = validate_config(invalid_config1)
|
|
print(f" 필수 항목 누락 감지: {'✅ PASS' if not is_valid else '❌ FAIL'}")
|
|
if not is_valid:
|
|
print(f" 오류 메시지: {msg}")
|
|
|
|
# 잘못된 설정 2: 범위 오류
|
|
invalid_config2 = {
|
|
"buy_check_interval_minutes": 0, # 1 미만
|
|
"stop_loss_check_interval_minutes": 60,
|
|
"profit_taking_check_interval_minutes": 240,
|
|
"dry_run": True,
|
|
"auto_trade": {},
|
|
}
|
|
|
|
is_valid, msg = validate_config(invalid_config2)
|
|
print(f" 범위 오류 감지: {'✅ PASS' if not is_valid else '❌ FAIL'}")
|
|
if not is_valid:
|
|
print(f" 오류 메시지: {msg}")
|
|
|
|
return True
|
|
|
|
|
|
def test_rebuy_prevention():
|
|
"""재매수 방지 기능 테스트"""
|
|
print("\n" + "=" * 70)
|
|
print("TEST 3: 재매수 방지 기능")
|
|
print("=" * 70)
|
|
|
|
import os
|
|
|
|
from src.common import can_buy, record_sell
|
|
|
|
test_symbol = "KRW-TEST"
|
|
|
|
# 테스트 파일 정리
|
|
from src.common import RECENT_SELLS_FILE
|
|
|
|
if os.path.exists(RECENT_SELLS_FILE):
|
|
os.remove(RECENT_SELLS_FILE)
|
|
|
|
# 초기 상태: 매수 가능
|
|
result = can_buy(test_symbol, cooldown_hours=24)
|
|
print(f" 초기 상태 (매수 가능): {'✅ PASS' if result else '❌ FAIL'}")
|
|
|
|
# 매도 기록
|
|
record_sell(test_symbol)
|
|
print(" 매도 기록 저장 완료")
|
|
|
|
# 매도 직후: 매수 불가
|
|
result = can_buy(test_symbol, cooldown_hours=24)
|
|
print(f" 매도 직후 (매수 불가): {'✅ PASS' if not result else '❌ FAIL'}")
|
|
|
|
# 짧은 쿨다운으로 테스트 (1초)
|
|
time.sleep(2)
|
|
result = can_buy(test_symbol, cooldown_hours=1 / 3600) # 1초를 시간으로 변환
|
|
print(f" 쿨다운 경과 후 (매수 가능): {'✅ PASS' if result else '❌ FAIL'}")
|
|
|
|
# 테스트 파일 정리
|
|
if os.path.exists(RECENT_SELLS_FILE):
|
|
os.remove(RECENT_SELLS_FILE)
|
|
|
|
return True
|
|
|
|
|
|
def test_max_price_update():
|
|
"""최고가 갱신 기능 테스트"""
|
|
print("\n" + "=" * 70)
|
|
print("TEST 4: 최고가 갱신 기능")
|
|
print("=" * 70)
|
|
|
|
import os
|
|
|
|
from src.holdings import load_holdings, save_holdings, update_max_price
|
|
|
|
test_symbol = "KRW-TEST"
|
|
test_holdings_file = "data/test_holdings.json"
|
|
|
|
# 테스트 데이터 준비
|
|
initial_holdings = {test_symbol: {"buy_price": 10000, "amount": 1.0, "max_price": 10500}}
|
|
|
|
save_holdings(initial_holdings, test_holdings_file)
|
|
print(" 초기 최고가: 10500")
|
|
|
|
# 더 높은 가격으로 갱신
|
|
update_max_price(test_symbol, 11000, test_holdings_file)
|
|
holdings = load_holdings(test_holdings_file)
|
|
new_max = holdings[test_symbol]["max_price"]
|
|
print(f" 11000으로 갱신 후: {new_max}")
|
|
print(f" 갱신 성공: {'✅ PASS' if new_max == 11000 else '❌ FAIL'}")
|
|
|
|
# 더 낮은 가격으로 시도 (갱신 안 됨)
|
|
update_max_price(test_symbol, 10800, test_holdings_file)
|
|
holdings = load_holdings(test_holdings_file)
|
|
max_after_lower = holdings[test_symbol]["max_price"]
|
|
print(f" 10800으로 시도 후: {max_after_lower}")
|
|
print(f" 갱신 안 됨 (유지): {'✅ PASS' if max_after_lower == 11000 else '❌ FAIL'}")
|
|
|
|
# 테스트 파일 정리
|
|
if os.path.exists(test_holdings_file):
|
|
os.remove(test_holdings_file)
|
|
|
|
return True
|
|
|
|
|
|
def test_telegram_message_split():
|
|
"""Telegram 메시지 분할 기능 테스트"""
|
|
print("\n" + "=" * 70)
|
|
print("TEST 5: Telegram 메시지 분할 (DRY RUN)")
|
|
print("=" * 70)
|
|
|
|
# 실제 전송은 하지 않고 로직만 테스트
|
|
long_message = "A" * 5000 # 4000자 초과
|
|
|
|
max_length = 4000
|
|
chunks = [long_message[i : i + max_length] for i in range(0, len(long_message), max_length)]
|
|
|
|
print(f" 원본 메시지 길이: {len(long_message)}자")
|
|
print(f" 분할 개수: {len(chunks)}개")
|
|
print(f" 각 청크 길이: {[len(c) for c in chunks]}")
|
|
|
|
if len(chunks) == 2 and len(chunks[0]) == 4000 and len(chunks[1]) == 1000:
|
|
print("✅ PASS: 메시지가 올바르게 분할되었습니다")
|
|
return True
|
|
else:
|
|
print("❌ FAIL: 메시지 분할 오류")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""모든 테스트 실행"""
|
|
print("\n" + "🔍 코드 리뷰 개선사항 검증 시작")
|
|
print("=" * 70)
|
|
|
|
results = []
|
|
|
|
try:
|
|
results.append(("Rate Limiter", test_rate_limiter()))
|
|
except Exception as e:
|
|
print(f"❌ Rate Limiter 테스트 실패: {e}")
|
|
results.append(("Rate Limiter", False))
|
|
|
|
try:
|
|
results.append(("Config Validation", test_config_validation()))
|
|
except Exception as e:
|
|
print(f"❌ 설정 검증 테스트 실패: {e}")
|
|
results.append(("Config Validation", False))
|
|
|
|
try:
|
|
results.append(("Rebuy Prevention", test_rebuy_prevention()))
|
|
except Exception as e:
|
|
print(f"❌ 재매수 방지 테스트 실패: {e}")
|
|
results.append(("Rebuy Prevention", False))
|
|
|
|
try:
|
|
results.append(("Max Price Update", test_max_price_update()))
|
|
except Exception as e:
|
|
print(f"❌ 최고가 갱신 테스트 실패: {e}")
|
|
results.append(("Max Price Update", False))
|
|
|
|
try:
|
|
results.append(("Message Split", test_telegram_message_split()))
|
|
except Exception as e:
|
|
print(f"❌ 메시지 분할 테스트 실패: {e}")
|
|
results.append(("Message Split", False))
|
|
|
|
# 결과 요약
|
|
print("\n" + "=" * 70)
|
|
print("📊 테스트 결과 요약")
|
|
print("=" * 70)
|
|
|
|
passed = sum(1 for _, result in results if result)
|
|
total = len(results)
|
|
|
|
for name, result in results:
|
|
status = "✅ PASS" if result else "❌ FAIL"
|
|
print(f" {name}: {status}")
|
|
|
|
print("\n" + f"총 {passed}/{total} 테스트 통과")
|
|
|
|
if passed == total:
|
|
print("\n🎉 모든 개선사항이 정상적으로 구현되었습니다!")
|
|
return 0
|
|
else:
|
|
print("\n⚠️ 일부 테스트가 실패했습니다. 로그를 확인하세요.")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|