143 lines
5.2 KiB
Python
143 lines
5.2 KiB
Python
"""
|
||
📁 sightray/sightray.py
|
||
|
||
SightRay 전체 통합 실행 스크립트
|
||
- 사용자가 종목 코드, 날짜, 자본금만 입력하면 전체 파이프라인 자동 실행
|
||
- 수집 → 학습 → 분석 → 리스크 평가 → 전략 실행까지 한 번에 완료
|
||
- 실행 전 필요한 라이브러리 설치 여부 확인
|
||
"""
|
||
|
||
import os
|
||
import importlib.util
|
||
import subprocess
|
||
import sys
|
||
import glob
|
||
from dotenv import load_dotenv, set_key
|
||
|
||
# 필요한 패키지 목록 (import명: 설치명)
|
||
REQUIRED_PACKAGES = {
|
||
"pandas": "pandas",
|
||
"requests": "requests",
|
||
"xgboost": "xgboost",
|
||
"dotenv": "python-dotenv"
|
||
}
|
||
|
||
def check_and_install_packages():
|
||
missing = []
|
||
for import_name, pip_name in REQUIRED_PACKAGES.items():
|
||
if importlib.util.find_spec(import_name) is None:
|
||
missing.append(pip_name)
|
||
if not missing:
|
||
return
|
||
print(f"❗ 필수 라이브러리 누락: {', '.join(missing)}")
|
||
confirm = input("자동으로 설치할까요? (y/n): ").strip().lower()
|
||
if confirm == 'y':
|
||
subprocess.check_call([sys.executable, "-m", "pip", "install"] + missing)
|
||
print("✅ 설치 완료. 다시 실행해 주세요.")
|
||
sys.exit()
|
||
else:
|
||
print("❌ 설치되지 않았습니다. 수동으로 설치 후 다시 실행하세요.")
|
||
sys.exit()
|
||
|
||
# 라이브러리 설치 여부 확인
|
||
check_and_install_packages()
|
||
|
||
# API 키 확인 및 설정
|
||
|
||
# 🌐 .env 파일 체크 및 API 키 확인/생성
|
||
if not os.path.exists(".env"):
|
||
with open(".env", "w") as f:
|
||
f.write("# SightRay 환경설정 파일\n")
|
||
load_dotenv()
|
||
api_key = os.getenv("POLYGON_API_KEY")
|
||
if not api_key or '--reset-api' in sys.argv:
|
||
print("🔐 Polygon API 키가 설정되어 있지 않거나 초기화 요청이 감지되었습니다.")
|
||
new_key = input("Polygon API 키를 입력하세요: ").strip()
|
||
if new_key:
|
||
set_key(".env", "POLYGON_API_KEY", new_key)
|
||
print("✅ API 키가 .env 파일에 저장되었습니다.")
|
||
else:
|
||
raise ValueError("API 키가 입력되지 않았습니다.")
|
||
print("🔑 Polygon API 키가 설정되어 있지 않거나 초기화 요청이 감지되었습니다.")
|
||
new_key = input("Polygon API 키를 입력하세요: ").strip()
|
||
if new_key:
|
||
print("✅ API 키가 .env 파일에 저장되었습니다.")
|
||
|
||
# 엔진 import
|
||
from data_collection_engine.engine import DataCollectionEngine
|
||
from data_analysis_engine.analyzer import analyze_stocks_pipeline
|
||
from data_analysis_engine.train_model import train_model
|
||
from data_analysis_engine.dataset_builder import build_dataset
|
||
from risk_manage_engine.evaluate import evaluate_predictions
|
||
from strategy_engine.run_strategy import run_strategy
|
||
|
||
# (사용되지 않지만 보관된 build_dataset 예시)
|
||
def build_dataset(df):
|
||
df = df.copy()
|
||
df['target'] = (df['close'].shift(-1) > df['close']).astype(int)
|
||
df.dropna(inplace=True)
|
||
X = df[['open', 'high', 'low', 'close', 'volume']]
|
||
y = df['target']
|
||
return X, y
|
||
|
||
def run_sightray(symbol: str, start: str, end: str, capital: float, strategy: str = "adjusted"):
|
||
# 파일 경로 설정
|
||
data_dir = "data"
|
||
ohlcv_path = os.path.join(data_dir, f"{symbol}_ohlcv.csv")
|
||
prediction_path = "prediction_result.csv"
|
||
risk_path = "risk_filtered_result.csv"
|
||
output_path = "final_strategy_result.csv"
|
||
|
||
# 최신 모델 자동 탐색
|
||
model_dir = "data_analysis_engine/models"
|
||
model_list = sorted(glob.glob(f"{model_dir}/model_*.json"), reverse=True)
|
||
if not model_list:
|
||
raise FileNotFoundError("모델 파일이 존재하지 않습니다.")
|
||
model_path = model_list[0]
|
||
|
||
# 1️⃣ 데이터 수집
|
||
print("[1단계] 데이터 수집 중...")
|
||
engine = DataCollectionEngine(data_dir=data_dir)
|
||
engine.collect(symbol, start, end)
|
||
|
||
# 2️⃣ 모델 학습
|
||
print("[2단계] 분석 모델 학습 중...")
|
||
train_model(data_dir, output_model_path=model_path)
|
||
|
||
# 3️⃣ 분석 예측
|
||
print("[3단계] 예측 실행 중...")
|
||
analyze_stocks_pipeline(
|
||
cds_dir=data_dir,
|
||
model_path=model_path,
|
||
top_n=10,
|
||
output_path=prediction_path
|
||
)
|
||
|
||
# 4️⃣ 리스크 평가
|
||
print("[4단계] 리스크 평가 실행 중...")
|
||
evaluate_predictions(prediction_path, data_dir, output_path=risk_path)
|
||
|
||
# 5️⃣ 전략 실행
|
||
print("[5단계] 전략 생성 및 포지션 설정 중...")
|
||
run_strategy(
|
||
risk_filtered_path=risk_path,
|
||
total_capital=capital,
|
||
strategy=strategy,
|
||
output_path=output_path
|
||
)
|
||
|
||
print("\n✅ SightRay 전체 파이프라인 실행 완료!")
|
||
print(f"→ 최종 전략 결과: {output_path}")
|
||
|
||
if __name__ == "__main__":
|
||
print("SightRay 실행을 시작합니다. 다음 정보를 입력해 주세요:")
|
||
symbol = input("종목 코드 (예: AAPL, TSLA): ").strip().upper()
|
||
start = input("시작일 (YYYY-MM-DD): ").strip()
|
||
end = input("종료일 (YYYY-MM-DD): ").strip()
|
||
capital = float(input("총 투자 자본 (예: 1000000): ").strip())
|
||
strategy = input("전략 템플릿 (adjusted / basic) [기본: adjusted]: ").strip().lower()
|
||
if strategy not in ["basic", "adjusted"]:
|
||
strategy = "adjusted"
|
||
|
||
run_sightray(symbol, start, end, capital, strategy)
|