Chess Opening Recommendation with NeuMF model (1)

 

최근 머신러닝 공부를 위해 미니 프로젝트로 포트폴리오 웹사이트를 만들어 보고 있다. 오랜만에 진행하는 개인 프로젝트이기도 하고 개발 실력을 좀 기르고 싶어서 프로젝트 메뉴를 만들고, 각 메뉴를 클릭하면 각각의 서브 페이지로 이동하는 방식으로 규모를 좀 키워서 작업하고 있다. 프론트로는 예전에 몇번 다뤄본 react를 사용했고 백엔드는 처음 써보는 fastapi를 이용했다. LLM이 없던 시절에 비하면 확실히 새로운 것을 배우기가 엄청 편해진 것 같다.

 

claude가 그려준 구조도

 

MNIST 숫자 예측 기능은 테스트용으로 넣어봤고, 처음으로 떠올린 프로젝트는 체스 오프닝 추천이었다. 개발 과정에서 공부한 것들과 느낀 점들을 가볍게 정리해 보고자 한다.

 

 

데이터 수집

체스 오프닝 데이터는 내가 다뤄본 데이터 중 가장 까다로운 유형의 데이터였다. 먼저 오프닝 이름과 ECO, 기보(PGN)를 포함한 목록이 필요했기 때문에, https://github.com/hayatbiralem/eco.json 에서 해당 데이터를 얻었다. (추후에 리체스 공식 레포 https://github.com/lichess-org/chess-openings 가 있는 것을 뒤늦게 발견했다..)

 

먼저 오프닝들을 마지막 수를 기준으로 백/흑으로 나눠 주었다. 간단히 시각화를 해 보니, 지나치게 마이너한 오프닝들도 있었고, 승률이 원사이드해 사실상 게임이 끝났다고 여겨지는 오프닝들도 있었다. 또한 7수 이상의 오프닝은 추천하는 의미가 크게 없다고 여겨졌다. 이러한 이상치들은 향후 추천에 영향을 끼칠 수 있을 것이라고 판단되어 필터링을 거쳤다.

 

# filter opening: ~6 moves, >50000 games, 30~70% winrate 
def select_opening(moves, opening_original, fn):
    for i, (_, m) in tqdm(enumerate(moves)):
        fen = moves_to_fen[m]
        if opening_original[fen]["move_num"] > 6: continue

        data = get_data(f"https://explorer.lichess.org/lichess?fen={fen}&topGames=0&recentGames=0&since=2015-01&speeds=blitz,rapid,classical&ratings=1400,1600,1800,2000,2200,2500")
        games = data["white"]+data["draws"]+data["black"]

        if games >= 50000 and 0.3 <= data["white"]/games <= 0.7 and 0.3 <= data["black"]/games <= 0.7:
            opening[fen] = opening_original[fen]
            opening[fen]["id"] = i

 

 

그 다음으로는 분석을 위해 lichess API로 통계 자료를 수집해야 했다. 각 오프닝마다 레이팅 구간(~1000, 1200, ... , 2200, 2500~) 및 시간 제한(bullet, blitz, rapid, classical) 별로 api request를 보내 데이터를 수집했다. 데이터의 양은 차고 넘치도록 많기 때문에 이상치를 줄이는 방향이 낫겠다고 생각을 했다. 그래서 2015년 이후 데이터만 사용했으며, 레이팅별 자료를 수집할 때는 blitz/rapid/classical 게임에 한해서, 시간 제한별 자료를 수집할 때는 rating 1400+ 이상으로 필터링해서 수집했다. 약 8시간 정도 걸려서 백 승/흑 승/무승부 게임 수, 오프닝 사용자의 평균 레이팅 자료를 구할 수 있었다. 초반에 개발 방향을 수정하는 과정에서, 데이터를 다시 수집해야 하는 일이 있었다. 이런 사고를 막기 위해서는 한 번 수집할 때 최대로 수집 가능한 데이터들을 raw 형태를 최대한 보존해서 수집해야 할 것이다.

 

with open(f"./backend/db/chess/db_{fn}_selected.csv", "a+", newline='', encoding='utf-8') as f:
    writer = csv.writer(f)
    for i, fen in tqdm(enumerate(opening)):
        if not (start<=i): continue

        # basic info
        line = [opening[fen]["id"], fen, opening[fen]["name"], opening[fen]["moves"], opening[fen]["eco"]]

        data = get_data(f"https://explorer.lichess.org/lichess?fen={fen}&topGames=0&recentGames=0&since=2015-01&speeds=blitz,rapid,classical&ratings=1400,1600,1800,2000,2200,2500")
        line += [data["white"], data["draws"], data["black"]]

        # avg rating
        for s in SPEEDS:   
            data = get_data(f"https://explorer.lichess.org/lichess?fen={fen}&topGames=0&recentGames=0&since=2015-01&speeds={s}")
            games = data["white"]+data["draws"]+data["black"]
            ar = 0
            for move in data["moves"]:
                ar += move["averageRating"]*(move["white"]+move["draws"]+move["black"])
            line.append(int(ar/games*10)/10)

        # rating
        for r in RATINGS:
            data = get_data(f"https://explorer.lichess.org/lichess?fen={fen}&topGames=0&recentGames=0&since=2015-01&speeds=blitz,rapid,classical&ratings={r}")
            line += [data["white"], data["draws"], data["black"]]

        # speed
        for s in SPEEDS:
            data = get_data(f"https://explorer.lichess.org/lichess?fen={fen}&topGames=0&recentGames=0&since=2015-01&speeds={s}&ratings=1400,1600,1800,2000,2200,2500")
            line += [data["white"], data["draws"], data["black"]]

        writer.writerow(line)
        f.flush()

 

 

 

데이터 가공

다음으로 수집한 데이터를 dataFrame으로 불러와 가공 과정을 거쳤다. 간단하게 승률 및 score%(승률+무승부율/2)를 구해 주고, 오프닝 세부 분석 페이지 및 추천 과정에서 사용할 만한 몇 가지 인사이트 지표들을 고민해 보았다. 첫 번째 지표로는 오프닝의 날카로움 정도를 측정하는 sharpness 지표를 고안했다. 무승부율을 사용하면 되겠다고 생각했는데, plot을 찍어보니 평균 레이팅에 따른 무승부율이 대체로 선형 관계에 있었다. 그래서 LinearRegression으로 적합해주고 회귀 직선과의 오차를 sharpness의 척도로 삼았다.

 

두 번째로 오프닝의 유명도를 popularity 지표로 나타내기로 했다. 이것도 naive하게 플레이된 게임 수를 사용하기는 무리가 있는데, 기본적으로 오프닝의 깊이(move 수)에 따라 게임 수가 기하급수적으로 줄어들기 때문이다. 그래서 move_num을 기준으로 그룹으로 묶어 주고, 각 그룹 내에서 게임 수 비율을 standardScaler로 표준화하여 사용하였다.

 

그 외에 레이팅별 데이터, 시간 제한별 데이터를 가지고 elo_sensivity랑 time_pressure_advantage 지표를 만들었다. elo_sensitivity는 [1200, 2200] 구간 데이터를 가지고 LinearRegression으로 구했고, time_pressure_advantage의 경우 게임 수 편차가 큰 관계로 단순히 (bullet, blitz 가중 평균) - (rapid, classical 가중 평균)으로 구했다. 다만 이들은 프론트에 display만 하고 실제 추천 시스템에 반영하지는 않았다.

 

# get sharpness
for s in SPEEDS:
    sharp_model = LinearRegression()
    sharp_model.fit(df[f"{s}_avg"].values.reshape(-1, 1), df["draws_rate"])
    df[f"{s}_sharpness"] = -(df["draws_rate"] - (sharp_model.coef_[0]*df[f"{s}_avg"] + sharp_model.intercept_))
    stats[f"{s}_sharpness_coef"] = [sharp_model.coef_[0], sharp_model.intercept_]
df["sharpness"] = (df["bullet_sharpness"] + df["blitz_sharpness"] + df["rapid_sharpness"] + df["classical_sharpness"]) / 4
df["sharpness"] = ss.fit_transform(df["sharpness"].values.reshape(-1, 1))

# get popularity
df["move_num"] = df["moves"].str.split('.').str[-2].str.strip().str.split(' ').str[-1]
df["games_by_move_num"] = df.groupby("move_num")["games"].transform('sum')
df["popularity"] = np.log(df['games'] / df['games_by_move_num'] + 1e-10)
scaler = lambda x: (x - x.mean()) / x.std()
df["popularity"] = df.groupby("move_num")["popularity"].transform(scaler)
df.drop(columns=['move_num', 'games_by_move_num'], inplace=True)

# get elo sensitivity
ES_TARGET_RATINGS = [1200, 1400, 1600, 1800, 2000, 2200]
ES_X = np.array(ES_TARGET_RATINGS).reshape(-1, 1)
rating_avg = [np.sum(df[f"{r}_score_rate"]*df[f"{r}_games"])/df[f"{r}_games"].sum() for r in ES_TARGET_RATINGS]
slopes = []
for _ , row in df.iterrows():
    y = [row[f"{r}_score_rate"] - rating_avg[i] for i, r in enumerate(ES_TARGET_RATINGS)]
    slopes.append(LinearRegression().fit(ES_X, y).coef_[0])
df["elo_sensitivity"] = slopes
df["elo_sensitivity"] = StandardScaler().fit_transform(df["elo_sensitivity"].values.reshape(-1, 1))

# get time pressure advantage
bullet_avg = np.sum(df["bullet_score_rate"]*df["bullet_games"])/df["bullet_games"].sum()
blitz_avg = np.sum(df["blitz_score_rate"]*df["blitz_games"])/df["blitz_games"].sum()
rapid_avg = np.sum(df["rapid_score_rate"]*df["rapid_games"])/df["rapid_games"].sum()
classical_avg = np.sum(df["classical_score_rate"]*df["classical_games"])/df["classical_games"].sum()
df["time_pressure_advantage"] = ((df["bullet_score_rate"]-bullet_avg)*df["bullet_games"] + (df["blitz_score_rate"]-blitz_avg)*df["blitz_games"])/(df["bullet_games"]+df["blitz_games"]) -\
                                ((df["rapid_score_rate"]-rapid_avg)*df["rapid_games"] + (df["classical_score_rate"]-classical_avg)*df["classical_games"])/(df["rapid_games"]+df["classical_games"])
df["time_pressure_advantage"] = StandardScaler().fit_transform(df["time_pressure_advantage"].values.reshape(-1, 1))