main.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from datetime import datetime
import json

import sys
import requests
import pandas as pd
from bs4 import BeautifulSoup
import re
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta
import numpy as np
from typing import List, Union, Dict
import yfinance as yf

#주식 종목 코드(code)에 대해서 최초 상장일(origintime)을 가져옴 , 언제부터 백테스팅 돌릴지 체크
#네이버 금융 주식 데이터 사용
def get_stock_origintime(code):
    try:
        url = "https://fchart.stock.naver.com/sise.nhn?symbol={}&timeframe=day&count=1&requestType=0".format(code)
        html = requests.get(url).text
        soup = BeautifulSoup(html, "lxml")
        origintime = soup.select_one("chartdata")['origintime']
        return origintime
    except Exception:
        raise ValueError(f"Stock code {code} not found")

#주식 종목 코드(code), 시작일, 종료일에 대해서 주식 data를 가져옴
#네이버 금융 주식 데이터 사용
def get_stock_data(code, from_date, to_date):
    try:
        from_date = str(from_date)
        to_date = str(to_date)
        count = (datetime.today() - datetime.strptime(from_date, "%Y%m%d")).days + 1

        stock_data = []
        url = "https://fchart.stock.naver.com/sise.nhn?symbol={}&timeframe=day&count={}&requestType=0".format(code, count)
        html = requests.get(url).text
        soup = BeautifulSoup(html, "lxml")
        data = soup.findAll('item')
        for row in data:
            daily_history = re.findall(r"[-+]?\d*\.\d+|\d+", str(row))
            #값이 설정한 기간 내에 존재한 다면
            if int(daily_history[0]) >= int(from_date) and int(daily_history[0]) <= int(to_date):
                daily_history[0] = datetime.strptime(daily_history[0], "%Y%m%d")
                daily_history[1] = float(daily_history[1])
                daily_history[2] = float(daily_history[2])
                daily_history[3] = float(daily_history[3])
                daily_history[4] = float(daily_history[4])
                daily_history[5] = float(daily_history[5])
                stock_data.append(daily_history)

        #pandas 이용을 위해 dataframe객체로 변환
        df = pd.DataFrame(stock_data, columns=['date', 'price', 'high', 'low', 'close', 'vol'])
        df.set_index(keys='date', inplace=True)
        return df
    except Exception:
        raise Exception(f"Failed to fetch data for stock code {code}")

# 해외 주식 데이터 가져오기 (Yahoo Finance 사용)
def get_stock_data_yahoo(code, from_date, to_date, exchange_rate=1350):
    try:
        df = yf.download(code, start=from_date, end=to_date)
        if df.empty:
            raise ValueError(f"No data found for stock code {code} in the given date range")
        
        df.reset_index(inplace=True)
        df['date'] = pd.to_datetime(df['Date'])
        df = df[['date', 'Open', 'High', 'Low', 'Close', 'Volume']].rename(
            columns={'Open': 'price', 'High': 'high', 'Low': 'low', 'Close': 'close', 'Volume': 'vol'}
        )
        df.set_index('date', inplace=True)
        # 고정 환율을 사용하여 가격을 원화로 변환
        df['close'] = df['close'] * exchange_rate
        df['price'] = df['price'] * exchange_rate
        df['high'] = df['high'] * exchange_rate
        df['low'] = df['low'] * exchange_rate
        
        return df
    
    except Exception as e:
        raise ValueError(f"Failed to fetch data for stock code {code} from Yahoo Finance: {str(e)}")

##############################
# 주식 매수 처리
##############################
#리밸런싱 용 코드. 비율에 맞춰 주식을 매수하거나 매도 처리
def buy_stock(money, stock_price, last_stock_num, stock_rate):
    if stock_price == 0:
        return money, 0, 0

    stock_num = money * stock_rate // stock_price
    stock_money = stock_num * stock_price
    if last_stock_num < stock_num:
        fee = 0.00015 # 매수 수수료, 토스 증권 기준
    else:
        fee = 0.000195 # 매도 수수료, 토스 증권 기준
    buy_sell_fee = abs(last_stock_num - stock_num) * stock_price * fee

    #돈 없으면 주식 갯수 조정
    while stock_num > 0 and money < (stock_money + buy_sell_fee):
        stock_num -= 1
        stock_money = stock_num * stock_price
        buy_sell_fee = abs(last_stock_num - stock_num) * stock_price * fee

    money -= (stock_money + buy_sell_fee)
    return money, stock_num, stock_money

#보유 자산에 현금 추가 되었을때 (안씀)
def buy_stock_more(money, stock_price, last_stock_num, stock_rate):
    if stock_price == 0:
        return money, 0, 0

    stock_num = money * stock_rate // stock_price
    stock_money = stock_num * stock_price
    if last_stock_num < stock_num:
        fee = 0.00015 # 매수 수수료 ,토스 증권 기준
    else:
        fee = 0.00195 # 매도 수수료, 토스 증권 기준
    buy_sell_fee = stock_num * stock_price * fee
    while stock_num > 0 and money < (stock_money + buy_sell_fee):
        stock_num -= 1
        stock_money = stock_num * stock_price
        buy_sell_fee = stock_num * stock_price * fee
    money -= (stock_money + buy_sell_fee)

    stock_num = stock_num + last_stock_num
    stock_money = stock_num * stock_price

    return money, stock_num, stock_money

##############################
# 주식명, 가격, 비율에 대해 비율 재조정
##############################
def get_ratio(names, prices, ratios):
    total_ratio = 0
    new_ratios = []
    for name in names:
        if prices[name] > 0:
            total_ratio += ratios[names.index(name)]
            new_ratios.append(ratios[names.index(name)])
        else:
            new_ratios.append(0)

    for i in range(len(new_ratios)):
        new_ratios[i] = round(new_ratios[i] * 1 / total_ratio, 2)

    return new_ratios

##############################
# 월말 데이터 추출
##############################
#월말 데이터 추출(왜 인지 resample을 'M'말고 'ME'로 잡으라고 나옴)
def get_month_end_data(df):
    df.index = pd.to_datetime(df.index)
    return df.resample('ME').last()

##############################
# 샤프 비율, 연간 표준편차, 연간 수익률 계산
##############################
#df와 무위험 이자율 데이터로 샤프비율,표준편차(std),연간 수익률 계산
def calculate_sharpe_ratio_and_std(df, risk_free_rate=0.03):
    df.index = pd.to_datetime(df.index)
    df['monthly_return'] = df['backtest'].pct_change().dropna()

    monthly_std_dev = df['monthly_return'].std()

    cumulative_return = df['backtest'].iloc[-1] / df['backtest'].iloc[0] - 1
    total_period_years = (df.index[-1] - df.index[0]).days / 365.25
    annual_return = (1 + cumulative_return) ** (1 / total_period_years) - 1
    annual_std_dev = monthly_std_dev * np.sqrt(12)
    sharpe_ratio = (annual_return - risk_free_rate) / annual_std_dev

    return round(sharpe_ratio, 2), round(annual_std_dev * 100, 2), round(annual_return, 2)

# 날짜 형식을 변환하는 함수
def convert_date_format(date_str):
    return datetime.strptime(date_str, "%Y%m%d").strftime("%Y-%m-%d")

# 국내 및 해외 주식 구분 및 데이터 가져오기
def get_stock_data_combined(code, from_date, to_date, is_foreign="false"):
    if is_foreign == "true":
        # 날짜 형식을 '%Y-%m-%d'로 변환
        from_date = convert_date_format(from_date)
        to_date = convert_date_format(to_date)
        return get_stock_data_yahoo(code, from_date, to_date) # 해외 주식의 경우
    else:
        return get_stock_data(code, from_date, to_date)

def back_test_portfolio(money: int, interval: int, start_day: str, end_day: str, stock_list, start_from_latest_stock: str):
    total_invest_money = money

    stock_code = []
    stock_name = []
    stock_ratio = []
    is_foreign_list = []

    for sss in stock_list:
        stock_code.append(sss[0])
        stock_name.append(sss[1])
        stock_ratio.append(sss[2])
        is_foreign_list.append(sss[3])  # 국내 주식은 "false", 해외 주식은 "true"

    if sum(stock_ratio) > 1:
        raise Exception("Sum of ratios is greater than 1.0")

    first_date = None
    for i in range(len(stock_code)):
        if is_foreign_list[i] == "false":  # 국내 주식만 체크
            org_time = get_stock_origintime(stock_code[i])
            if start_from_latest_stock == "true":
                if first_date is None or first_date < org_time:
                    first_date = org_time
            else:
                if first_date is None or first_date > org_time:
                    first_date = org_time
   
    if first_date is not None and int(first_date) > int(start_day):
        start_day = first_date
    # if first_date > start_day:
    #     start_day = first_date

    start_date = datetime.strptime(start_day, '%Y%m%d')
    cal_days = (datetime.strptime(end_day, "%Y%m%d") - start_date).days

    df = pd.DataFrame()

    for i in range(len(stock_code)):
        df_close = get_stock_data_combined(stock_code[i], start_day, end_day, is_foreign_list[i])['close']
        df_close = df_close.rename(stock_name[i])
        df_close.index = pd.to_datetime(df_close.index)
        df_close = get_month_end_data(df_close)
        df = pd.merge(df, df_close, how='outer', left_index=True, right_index=True)

    df.columns = stock_name
    df.fillna(0, inplace=True)

    if start_from_latest_stock == "true":
        latest_start_date = max(pd.to_datetime([get_stock_origintime(code) for code, is_foreign in zip(stock_code, is_foreign_list) if is_foreign == "false"]))
        # latest_start_date = max(pd.to_datetime([get_stock_origintime(code) for code in stock_code]))
        df = df[df.index >= latest_start_date]

    rebalanceing_date_list = []
    while start_date <= df.index[-1]:
        temp_date = start_date
        while temp_date not in df.index and temp_date < df.index[-1]:
            temp_date += timedelta(days=1)
        rebalanceing_date_list.append(temp_date)
        start_date += relativedelta(months=interval)

    backtest_index = []
    backtest_data = []

    etf_num = {etf: 0 for etf in stock_name}
    prices = {etf: 0 for etf in stock_name}
    etf_money = {etf: 0 for etf in stock_name}

    date_idx = 0
    for each in df.index:
        rebalnace_day = False
        if date_idx < len(rebalanceing_date_list) and each == rebalanceing_date_list[date_idx] and interval > 0:
            if (date_idx) % interval == 0:
                rebalnace_day = True
            date_idx += 1

        for stock in stock_name:
            prices[stock] = df[stock][each]
            if rebalnace_day is True:
                money += etf_num[stock] * prices[stock]

        recal_ratio = get_ratio(stock_name, prices, stock_ratio)

        total = 0
        cal = 0

        for stock in stock_name:
            try:
                if rebalnace_day is True:
                    money, etf_num[stock], etf_money[stock] = buy_stock(money, prices[stock], etf_num[stock], recal_ratio[stock_name.index(stock)]/((1-cal) if cal < 1 else 1))
                else:
                    money, etf_num[stock], etf_money[stock] = buy_stock_more(money, prices[stock], etf_num[stock], recal_ratio[stock_name.index(stock)]/((1-cal) if cal < 1 else 1))
            except Exception as e:
                print(e)

            if etf_num[stock] > 0:
                total += etf_money[stock]
                cal += recal_ratio[stock_name.index(stock)]

        total += money
        backtest_index.append(each)
        backtest_data.append(int(total)/total_invest_money)

    backtest_df = pd.DataFrame(backtest_data, index=backtest_index, columns=['backtest'])

    final_df = pd.concat([df, backtest_df], axis=1)

    for stock in stock_name:
        for pr in final_df[stock]:
            if pr > 0:
                final_df[stock] = final_df[stock] / pr
                break

    final_df.index = final_df.index.astype(str)
    final_df_dict = final_df.to_dict()

    sharpe_ratio, annual_std_dev, annual_return = calculate_sharpe_ratio_and_std(final_df)

    return final_df, final_df_dict, sharpe_ratio, annual_std_dev, annual_return, total

#최대 낙폭 계산.
def calculate_mdd(df):
    df['cumulative_max'] = df['backtest'].cummax()
    df['drawdown'] = df['backtest'] / df['cumulative_max'] - 1
    mdd = df['drawdown'].min()
    return mdd

def back_test(stock_info):
    #포트폴리오 객체
    portfolio = stock_info['portfolio']
    # 백테스팅 시점 결정
    # 값이 true일 경우 가장 늦게 상장 된 주식의 상장일 기준으로 백테스팅 시작.
    # 값이 false일 경우 가장 먼저 상장 된 주식의 상장일 기준으로 백테스팅 시작.
    # 현재는 false이므로 먼저 상장된 기준으로 백테스팅하고, 그때 당시 안되있으면 반영이 안됨. (값 0 으로 처리)
    start_from_latest_stock = stock_info['start_from_latest_stock']
    #주식 목록 (종목코드,주식이름,포트폴리오 비율)
    stock_list = portfolio['stock_list']
    #초기 투자 총 금액
    balance = portfolio['balance']
    #리밸런싱 단위 (개월) ex) interval=1 은 1달마다 리밸런싱을 함을 의미합니다.
    interval = portfolio['interval_month']
    #백테스팅 시작일자
    start_date = portfolio['start_date']
    #백테스팅 끝일자
    end_date = portfolio['end_date']
    #백테스트 실행
    final_df, final_df_dict, sharpe_ratio, annual_std_dev, annual_return, total_balance = back_test_portfolio(balance, interval, start_date, end_date, stock_list, start_from_latest_stock)

    #최대 낙폭 계산
    mdd = calculate_mdd(final_df)

    result = {'portfolio': final_df_dict, 'sharpe_ratio': sharpe_ratio, 'standard_deviation': annual_std_dev, 'annual_return': annual_return, 'total_balance': total_balance, 'mdd': mdd}

    return result


app = FastAPI()

# 입력을 위한 Pydantic 모델 정의
class Stock(BaseModel):
    code: str
    name: str
    ratio: float
    is_foreign: str # 해외 주식 여부를 "true" 또는 "false"로

class Portfolio(BaseModel):
    stock_list: List[Union[List[Union[str, float, str]], Stock]]
    balance: int
    interval_month: int
    start_date: str
    end_date: str

class BackTestRequest(BaseModel):
    start_from_latest_stock: str
    portfolio: Portfolio 

@app.post("/backtest/")
async def run_backtest(request: BackTestRequest):
    try:
        # 입력 데이터를 백테스트 함수가 처리할 수 있도록 변환
        stock_info = request.dict()
        result = back_test(stock_info)  # back_test는 원래 코드에서 가져옴
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

#####################
# 자산군 통합
#####################
class PortfolioResult(BaseModel):
    backtest: Dict[str, float]  # 날짜별 수익률
    sharpe_ratio: float
    standard_deviation: float
    annual_return: float
    total_balance: float
    mdd: float

class AssetGroup(BaseModel):
    name: str  # 자산군 이름 (예: 국내, 해외)
    ratio: float  # 자산군 비율 (예: 0.6, 0.4)
    portfolio_result: PortfolioResult  # 백테스트 결과 (각 자산군에서 얻은 데이터)

class IntegratedPortfolioRequest(BaseModel):
    asset_groups: List[AssetGroup]  # 자산군 리스트

# 통합 백테스트 계산 함수
def calculate_weighted_results(asset_groups: List[AssetGroup]):
    weighted_portfolio = pd.DataFrame()

    for asset_group in asset_groups:
        try:
            # 'backtest' 존재 여부 확인 및 접근
            if hasattr(asset_group.portfolio_result, 'backtest'):
                portfolio_data = pd.Series(asset_group.portfolio_result.backtest)
                weighted_data = portfolio_data * asset_group.ratio
            else:
                raise ValueError(f"Missing 'backtest' in portfolio_result for asset group {asset_group.name}")

            if weighted_portfolio.empty:
                weighted_portfolio = weighted_data
            else:
                weighted_portfolio = weighted_portfolio.add(weighted_data, fill_value=0)

        except Exception as e:
            print(f"Error processing asset group {asset_group.name}: {str(e)}")
            raise e

    # 통합 포트폴리오 계산
    weighted_portfolio_df = weighted_portfolio.to_frame(name="backtest")
    sharpe_ratio, annual_std_dev, annual_return = calculate_sharpe_ratio_and_std(weighted_portfolio_df)
    mdd = calculate_mdd(weighted_portfolio_df)

    return {
        'integrated_portfolio': {
            'backtest': weighted_portfolio.to_dict(),
            'sharpe_ratio': sharpe_ratio,
            'standard_deviation': annual_std_dev,
            'annual_return': annual_return,
            'mdd': mdd
        }
    }

@app.post("/backtest/integrated")
async def run_integrated_backtest(request: IntegratedPortfolioRequest):
    try:
        # 입력 데이터를 백테스트 함수가 처리할 수 있도록 변환
        asset_groups = request.asset_groups
        
        # 자산군 별 백테스트 결과를 기반으로 통합 백테스트 계산
        result = calculate_weighted_results(asset_groups)
        return result
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    

# 만약 해당 코드가 main.py로 저장되었다면, 다음 명령어로 실행합니다.
# uvicorn main:app --reload