<aside> β ꡬλ μ μ¬λ¬λΆ, μλ μμλλ‘ νμΌμ λ§λ€κ³ μ€νλ§ νμλ©΄ λ―Έκ΅ μμ₯ μ 체λ₯Ό μ€μΊνλ AI μμ΄μ νΈκ° κ°λλ©λλ€.
</aside>
<b>1λ¨κ³:</b> λμΌν ν΄λμ μλ μ 곡λ 3κ°μ νμ΄μ¬ νμΌ(data_loader.py, analysis_core.py, nexus_run.py)μ λ§λλλ€.\\n<b>2λ¨κ³:</b> nexus_run.py μλ¨μ CAPITAL_KRWμ λ³ΈμΈμ ν¬μκΈμ μ
λ ₯ν©λλ€.\\n<b>3λ¨κ³:</b> ν°λ―Έλ(CMD)μμ python nexus_run.pyλ₯Ό μ€νν©λλ€.
nexus_run.py νμΌμ 9νμ μλ target_universe λ³μλ₯Ό μμ νμ¬ λΆμ λμμ λ°κΏ μ μμ΅λλ€.
target_universe = "nasdaq100" # λμ€λ₯ 100λ§ λΆμ
target_universe = "sp500" # S&P 500 μ μ μ‘°μ¬
target_universe = "smallcap" # μ€λͺ°μΊ‘(μ£Όλμ£Όκ΅°) λΆμ
target_universe = "all" # λμ€λ₯+S&P+μ€λͺ°μΊ‘ ν΅ν© λΆμ (κ°λ ₯ μΆμ²!)
# data_loader.py
import yfinance as yf
import pandas as pd
from typing import List, Optional
def get_market_universe(index_type: str = "nasdaq100") -> List[str]:
"""μ νν μ§μμ ν°μ»€ λͺ©λ‘ μλ μμ§ (all μ ν μ ν΅ν© μμ§)"""
try:
tickers = []
if index_type in ["nasdaq100", "all"]:
url = "<https://en.wikipedia.org/wiki/Nasdaq-100>"
tickers += pd.read_html(url)[4]['Ticker'].tolist()
if index_type in ["sp500", "all"]:
url = "<https://en.wikipedia.org/wiki/List_of_S%26P_500_companies>"
tickers += pd.read_html(url)[0]['Symbol'].tolist()
if index_type in ["smallcap", "all"]:
tickers += ["CIFR", "MRAM", "INOD", "MSTR", "CLSK", "MARA", "RIOT", "WULF"]
return list(set(tickers)) # μ€λ³΅ μ κ±°
except Exception as e:
print(f"Error fetching universe: {e}")
return ["TSLA", "NVDA", "AVGO", "PLTR"]
def fetch_historical_data(ticker: str, period="1y"):
data = yf.download(ticker, period=period, progress=False)
if data.empty: return None
if isinstance(data.columns, pd.MultiIndex):
data.columns = ['_'.join(col).strip() for col in data.columns]
return data
# analysis_core.py
import pandas as pd
import numpy as np
def analyze_v6_vcp_quantitative(data):
"""RS μ μ λ° VCP(λ³λμ± μμΆ) λΆμ"""
if len(data) < 100: return 0, 0
close = data['Close']
rs_score = (close.iloc[-1] / close.iloc[-63]) - 1 # 3κ°μ μμ΅λ₯ κΈ°λ° RS
high_low = data['High'] - data['Low']
high_close = np.abs(data['High'] - close.shift(1))
low_close = np.abs(data['Low'] - close.shift(1))
atr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1).rolling(14).mean().iloc[-1]
return rs_score, atr
def calculate_v7_sizing(price, atr, risk_amt, cap_amt):
"""μ΄μ€ 리μ€ν¬ κ΄λ¦¬ μ¬μ΄μ§"""
stop_dist = max(atr * 1.5, price * 0.05)
shares = min(risk_amt / stop_dist, cap_amt / price)
return int(shares), price - stop_dist
# nexus_run.py
from data_loader import get_market_universe, fetch_historical_data
from analysis_core import analyze_v6_vcp_quantitative, calculate_v7_sizing
import pandas as pd
from datetime import datetime
# [μ€μ ] λ³ΈμΈμ μν©μ λ§κ² μμ νμΈμ
CAPITAL_KRW = 10000000 # ν¬μ μκΈ (KRW)
target_universe = "all" # nasdaq100, sp500, smallcap, all μ€ μ ν
FX_RATE = 1350
risk_usd = (CAPITAL_KRW * 0.01) / FX_RATE
cap_usd = (CAPITAL_KRW * 0.25) / FX_RATE
if __name__ == "__main__":
print(f"π Nexus Agent V5.5 κ°λ... μ λλ²μ€: {target_universe}")
tickers = get_market_universe(target_universe)
results = []
for t in tickers:
df = fetch_historical_data(t)
if df is None: continue
rs, atr = analyze_v6_vcp_quantitative(df)
if rs < 0.2: continue # κ°ν μ’
λͺ©λ§ μ λ³
price = df['Close'].iloc[-1]
shares, sl = calculate_v7_sizing(price, atr, risk_usd, cap_usd)
if shares > 0:
results.append([t, round(rs, 2), price, shares, round(sl, 2)])
print(f" [λ°κ΅΄] {t} | RS: {rs:.2f} | μΆμ²: {shares}μ£Ό")
report = pd.DataFrame(results, columns=['Ticker', 'RS', 'Price', 'Shares', 'StopLoss'])
print("\n" + "="*50 + "\nμ΅μ’
λΆμ 리ν¬νΈ\n" + "="*50)
print(report.sort_values(by='RS', ascending=False).to_markdown(index=False))