분석.csv → 판단 프로그램 만들기

score.py

0.01MB

 

import os
import glob
import numpy as np
import pandas as pd


# ======================================================
# 폴더 설정
# ======================================================

INPUT_FOLDER = "Output"
OUTPUT_FOLDER = "Output"

os.makedirs(OUTPUT_FOLDER, exist_ok=True)


# ======================================================
# 골든크로스
# ======================================================

def detect_golden_cross(df):

    cond = (
        (df["MACD"].shift(1) < df["Signal"].shift(1)) &
        (df["MACD"] >= df["Signal"])
    )

    df["GoldenCross"] = cond

    return df


# ======================================================
# 데드크로스
# ======================================================

def detect_dead_cross(df):

    cond = (
        (df["MACD"].shift(1) > df["Signal"].shift(1)) &
        (df["MACD"] <= df["Signal"])
    )

    df["DeadCross"] = cond

    return df


# ======================================================
# MACD 0선 돌파
# ======================================================

def detect_macd_zero(df):

    state = np.full(len(df), "", dtype=object)

    up = (
        (df["MACD"].shift(1) < 0) &
        (df["MACD"] >= 0)
    )

    down = (
        (df["MACD"].shift(1) > 0) &
        (df["MACD"] <= 0)
    )

    state[up] = "Zero Up"

    state[down] = "Zero Down"

    df["MACD_Zero"] = state

    return df


# ======================================================
# Histogram 증가 감소
# ======================================================

def detect_histogram(df):

    trend = np.full(len(df), "", dtype=object)

    increase = (
        df["Histogram"] >
        df["Histogram"].shift(1)
    )

    decrease = (
        df["Histogram"] <
        df["Histogram"].shift(1)
    )

    trend[increase] = "Increasing"

    trend[decrease] = "Decreasing"

    df["HistogramTrend"] = trend

    return df


# ======================================================
# RSI 상태
# ======================================================

def detect_rsi_state(df):

    state = np.full(len(df), "", dtype=object)

    state[df["RSI14"] >= 70] = "OverBought"

    state[df["RSI14"] <= 30] = "OverSold"

    normal = (
        (df["RSI14"] > 30) &
        (df["RSI14"] < 70)
    )

    state[normal] = "Normal"

    df["RSI_State"] = state

    return df


# ======================================================
# 기술 신호 계산
# ======================================================

def calculate_signal(df):

    df = detect_golden_cross(df)

    df = detect_dead_cross(df)

    df = detect_macd_zero(df)

    df = detect_histogram(df)

    df = detect_rsi_state(df)

    return df


###########################################################
# 이동평균선 상태
###########################################################

def detect_ma_state(df):

    state = np.full(len(df), "", dtype=object)

    # 정배열
    cond1 = (
        (df["MA5"] > df["MA20"]) &
        (df["MA20"] > df["MA60"])
    )

    # 역배열
    cond2 = (
        (df["MA5"] < df["MA20"]) &
        (df["MA20"] < df["MA60"])
    )

    state[cond1] = "Bull"

    state[cond2] = "Bear"

    state[state == ""] = "Mixed"

    df["MA_State"] = state

    return df


###########################################################
# 볼린저 위치
###########################################################

def detect_bb(df):

    state = np.full(len(df), "", dtype=object)

    state[df["종가"] > df["BB_Upper"]] = "Upper Break"

    state[df["종가"] < df["BB_Lower"]] = "Lower Break"

    inside = (
        (df["종가"] >= df["BB_Lower"]) &
        (df["종가"] <= df["BB_Upper"])
    )

    state[inside] = "Inside"

    df["BB_Position"] = state

    return df


###########################################################
# 거래량
###########################################################

def detect_volume(df):

    state = np.full(len(df), "", dtype=object)

    high = df["거래량"] > df["VOL_MA20"]

    low = df["거래량"] <= df["VOL_MA20"]

    state[high] = "High"

    state[low] = "Low"

    df["Volume_State"] = state

    return df


###########################################################
# 매수점수
###########################################################

def calculate_buy_score(df):

    score = np.zeros(len(df))

    score += np.where(df["GoldenCross"],20,0)

    score += np.where(df["MACD"]>0,10,0)

    score += np.where(df["HistogramTrend"]=="Increasing",10,0)

    score += np.where(df["MA_State"]=="Bull",15,0)

    score += np.where(df["종가"]>df["MA120"],10,0)

    score += np.where(df["종가"]>df["MA240"],5,0)

    score += np.where(
        (df["RSI14"]>=40)&(df["RSI14"]<=60),
        10,
        0
    )

    score += np.where(df["RSI14"]<30,15,0)

    score += np.where(df["Volume_State"]=="High",5,0)

    score=np.clip(score,0,100)

    df["BuyScore"]=score.astype(int)

    return df


###########################################################
# 매도점수
###########################################################

def calculate_sell_score(df):

    score=np.zeros(len(df))

    score += np.where(df["DeadCross"],20,0)

    score += np.where(df["MACD"]<0,10,0)

    score += np.where(df["HistogramTrend"]=="Decreasing",10,0)

    score += np.where(df["MA_State"]=="Bear",15,0)

    score += np.where(df["종가"]<df["MA120"],10,0)

    score += np.where(df["종가"]<df["MA240"],5,0)

    score += np.where(df["RSI14"]>70,15,0)

    score += np.where(df["Volume_State"]=="Low",5,0)

    score += np.where(df["BB_Position"]=="Upper Break",10,0)

    score=np.clip(score,0,100)

    df["SellScore"]=score.astype(int)

    return df


###########################################################
# 최종 의견
###########################################################

def opinion(row):

    b=row["BuyScore"]
    s=row["SellScore"]

    if b>=90:
        return "★★★★★ Strong Buy"

    elif b>=75:
        return "★★★★ Buy"

    elif s>=90:
        return "★★★★★ Strong Sell"

    elif s>=75:
        return "★★★★ Sell"

    else:
        return "★★★ Hold"


###########################################################
# 이유
###########################################################

def reason(row):

    txt=[]

    if row["GoldenCross"]:
        txt.append("GoldenCross")

    if row["DeadCross"]:
        txt.append("DeadCross")

    if row["MACD_Zero"]=="Zero Up":
        txt.append("MACD Zero Up")

    if row["HistogramTrend"]=="Increasing":
        txt.append("Histogram Up")

    if row["HistogramTrend"]=="Decreasing":
        txt.append("Histogram Down")

    if row["MA_State"]=="Bull":
        txt.append("Bull Trend")

    if row["MA_State"]=="Bear":
        txt.append("Bear Trend")

    if row["RSI_State"]=="OverBought":
        txt.append("RSI High")

    if row["RSI_State"]=="OverSold":
        txt.append("RSI Low")

    return ", ".join(txt)


###########################################################
# Main
###########################################################

def process(df):

    df=calculate_signal(df)

    df=detect_ma_state(df)

    df=detect_bb(df)

    df=detect_volume(df)

    df=calculate_buy_score(df)

    df=calculate_sell_score(df)

    df["Opinion"]=df.apply(opinion,axis=1)

    df["Reason"]=df.apply(reason,axis=1)

    return df


###########################################################
# 실행
###########################################################

def main():

    files=glob.glob(os.path.join(INPUT_FOLDER,"*_Analysis.csv"))

    if len(files)==0:

        print("Analysis 파일이 없습니다.")

        return

    for file in files:

        print("--------------------------------")

        print(os.path.basename(file))

        df=pd.read_csv(file,encoding="utf-8-sig")

        result=process(df)

        name=os.path.basename(file).replace("_Analysis.csv","")

        save=os.path.join(
            OUTPUT_FOLDER,
            name+"_Score.csv"
        )

        result.to_csv(
            save,
            index=False,
            encoding="utf-8-sig"
        )

        print("저장 :",save)

    print()

    print("완료")


if __name__=="__main__":

    main()