e-stat APIとかRESAS APIとか,Web APIは便利ですよね。
今回はPythonでAPIを作る方法をメモしておくことにします。
APIで提供する機能は「機械学習による予測」としておきます。
今回は
- APIの仕様を決めて
- データを取得し
- 予測モデルを作り
- APIをローカルで動作させる
ところまでを書きます。デプロイについては今後勉強して書きます。
要件を決める
作るものをざっくり決めておきます。
土地総合情報システム | 国土交通省のデータを使って,不動産の価格予測を行うシステムを作ることにします。
条件
- 東京都の中古マンションのみを対象とする
- リノベーションされた物件は対象外とする(築年数の持つ意味が変わるため)
APIに送信する特徴量
以下の特徴量を送ることにします。
特徴量 |
説明 |
データ型 |
address |
市区町村レベルまでの住所 |
string |
area |
専有面積 |
int or float |
building_year |
竣工年 |
int or float |
入力(request)/出力(response)
例えば予測のリクエストを
{
"address": "東京都千代田区",
"area": 30,
"building_year": 2013
}
のようにして送ると,
{
"status": "OK",
"preidcted": 40000000
}
のような値を返すAPIとします。
予測モデルを作る
データの取得
土地総合情報システム | 国土交通省のAPIを使います。
import requests
import json
import pandas as pd
import os
url = "https://www.land.mlit.go.jp/webland/api/TradeListSearch"
payload = {"area": 13, "from": 20053, "to": 20193}
response = requests.get(url, params=payload)
data = json.loads(response.text)
df = pd.DataFrame(data["data"])
os.mkdir("input")
df.to_csv("input/raw.csv", index=False)
基礎的な前処理
まず基礎的な前処理を行い,APIで受け取るデータと同様の状況にしていきます。
import pandas as pd
df = pd.read_csv("input/raw.csv")
is_mansion = df["Type"] == "中古マンション等"
df = df.loc[is_mansion, :]
is_not_renovated = df["Renovation"] != "改装済"
df = df.loc[is_not_renovated, :]
df = df.rename(columns={"TradePrice": "price", "Area": "area"})
df["address"] = df["Prefecture"] + df["Municipality"]
years = df["BuildingYear"].str.extract(r"(?P<period>昭和|平成|令和)(?P<year>\d+)")
years["year"] = years["year"].astype(float)
years["period"] = years["period"].map({"昭和": 1925, "平成": 1988, "令和": 2019})
df["building_year"] = years["period"] + years["year"]
years = df["Period"].str.extract(r"(\d+)")[0]
zen2han = {"1": "1", "2": "2", "3": "3", "4": "4"}
quarters = df["Period"].str.extract(r"(\d四半期)")[0]\
.str.replace("四半期", "").map(zen2han).astype(int)
months = (quarters * 3 - 2).astype(str)
df["trade_date"] = pd.to_datetime(years + "-" + months)
cols = ["price", "address", "area", "building_year", "trade_date"]
df = df[cols].dropna()
df.to_csv("input/basic_data.csv", index=False)
こんな感じになります。
- priceは目的変数
- address, area, building_yearは予測のときはAPIの利用者が入力する
- trade_dateは予測のときは「APIが利用された日」を使う(システムが作る特徴量)
という想定です。
前処理関数の定義
addressは文字列,trade_dateはdatetimeか文字列なので,このまま機械学習モデルに入れるわけにはいきません。
今回はLightGBMを使うので完全にカテゴリカル変数であるaddressはLightGBM内でcategoricalにすればいいとしても,trade_dateは順序があるカテゴリカル変数なので数値にしたいところです。
そんなちょっとした特徴量の加工をするクラスを定義します1
from sklearn.base import BaseEstimator, TransformerMixin
import pandas as pd
import numpy as np
class skPlumberBase(BaseEstimator, TransformerMixin):
"""Pipelineに入れられるTransformerのベース"""
def __init__(self):
pass
def fit(self, X, y=None):
return self
def transform(self, X):
return self
class Date2Int(skPlumberBase):
def __init__(self, target_col):
self.target_col = target_col
def transform(self, X):
"""unix時間に変換する"""
dates = pd.to_datetime(X[self.target_col]).astype(np.int64) / 10**9
X[self.target_col] = dates.astype(int)
return X
class ToCategorical(skPlumberBase):
"""LightGBMにcategoryだと認識させるため,
カテゴリカル変数をpandas category型にする
"""
def __init__(self, target_col):
self.target_col = target_col
def transform(self, X):
X[self.target_col] = X[self.target_col].astype("category")
return X
Date2Intは
0 2019-07-01
1 2018-10-01
2 2018-07-01
3 2018-04-01
4 2018-04-01
...
137548 2008-01-01
137549 2007-10-01
137550 2007-10-01
137551 2007-07-01
137552 2007-04-01
Name: trade_date, Length: 137553, dtype: object
を
0 1561939200
1 1538352000
2 1530403200
3 1522540800
4 1522540800
...
137548 1199145600
137549 1191196800
137550 1191196800
137551 1183248000
137552 1175385600
Name: trade_date, Length: 137553, dtype: int32
のようにするものです。
unix時間にすると桁数が増えて使用メモリが増えて効率的ではない気がしますが,あくまで例ということで…
学習
前処理して,学習して,前処理パイプラインと予測モデルをpickleで保存します。
from sklearn.pipeline import Pipeline
from pipeline import Date2Int, ToCategorical
import pandas as pd
import pickle
import lightgbm as lgb
from sklearn.model_selection import train_test_split
df = pd.read_csv("input/basic_data.csv")
y = df["price"]
X = df.drop("price", axis=1)
preprocess = Pipeline(steps=[
("date_to_int", Date2Int(target_col="trade_date")),
("to_category", ToCategorical(target_col="address"))
], verbose=True)
X = preprocess.transform(X)
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42)
params = {
"n_estimators": 100_000,
"min_child_samples": 15,
"max_depth": 4,
"colsample_bytree": 0.7,
"random_state": 42
}
model = lgb.LGBMRegressor(**params)
model.fit(X_train, y_train,
eval_metric="rmse",
eval_set=[(X_test, y_test)],
early_stopping_rounds=100)
print("best scores:", dict(model.best_score_["valid_0"]))
pickle.dump(preprocess, open("preprocess.pkl", "wb"))
pickle.dump(model, open("model.pkl", "wb"))
testデータに対するRMSEが1950万円くらいあります。これほど少ない特徴量だとさすがにひどい精度になりますね。
best scores: {'rmse': 19500026.074094355, 'l2': 380251016890359.75}
テストを書く
作るものを決める
これから作るAPIは,予測したい物件の特徴量を
{
"address": "東京都千代田区",
"area": 30,
"building_year": 2013
}
のようにJSONにして送ると,
{
"status": "OK",
"preidcted": 40000000
}
のようなJSONの値を返すAPIとします。
テストを書く
import unittest
import requests
import json
class APITest(unittest.TestCase):
URL = "http://localhost:5000/api/predict"
DATA = {
"address": "東京都千代田区",
"area": 30,
"building_year": 2013
}
def test_normal_input(self):
response = requests.post(self.URL, json=self.DATA)
print(response.text)
result = json.loads(response.text)
self.assertEqual(response.status_code, 200)
self.assertEqual(result["status"], "OK")
self.assertTrue(0 <= result["predicted"])
if __name__ == "__main__":
unittest.main()
※あくまで例です。実際にちゃんと作るときはもっと沢山(正常系だけでなく異常系も)テストケースを作ります。
Flaskでアプリを作成
FlaskはPythonの軽量なWebフレームワーク(Webアプリを簡単に作ることができるライブラリ)で,極めて少ないコード量でアプリを作ることができます。
「機械学習モデルを動かすだけ」みたいな単純な動作をするAPIには最適なフレームワークです。
以下のように書いていきます。
from flask import Flask, request, jsonify, abort
import pandas as pd
import pickle
from datetime import datetime
import sys
sys.path.append("./model")
app = Flask(__name__)
preprocess = pickle.load(open("model/preprocess.pkl", "rb"))
model = pickle.load(open("model/model.pkl", "rb"))
@app.route('/api/predict', methods=["POST"])
def predict():
"""/api/predict にPOSTリクエストされたら予測値を返す関数"""
try:
X = pd.DataFrame(request.json, index=[0])
X["trade_date"] = datetime.now()
X = preprocess.transform(X)
y_pred = model.predict(X, num_iteration=model.best_iteration_)
response = {"status": "OK", "predicted": y_pred[0]}
return jsonify(response), 200
except Exception as e:
print(e)
abort(400)
@app.errorhandler(400)
def error_handler(error):
"""abort(400) した時のレスポンス"""
response = {"status": "Error", "message": "Invalid Parameters"}
return jsonify(response), error.code
if __name__ == "__main__":
app.run(debug=True)
前節で書いたテストを走らせるとこうなります
$ python3 api_test.py
{
"predicted": 45833222.1903707,
"status": "OK"
}
.
----------------------------------------------------------------------
Ran 1 test in 0.015s
OK
期待通り,predictedとstatusが返っているようです。
今回のコードをまとめたGitHubリポジトリはこちらになります:
github.com