Pythonでシフトを自動作成するアプリを作成、運用した話

背景

現在焼肉屋でアルバイトをしている大学生です。
バイトのシフトの管理は完全に学生に任されています。
手書き入力の紙提出でなおかつ作成者は経験と勘で組まなければなりませんでした。
そこでPythonとかで解決できないかと考え、アプリ作成に至りました。

解決すべき問題

• 紙提出をやめてネットでの提出にする。

• シフト作成にかける時間を減らしつつそれなりみんなが満足するシフトを作る。

の２つでした。

アプローチ

まず考えついたのは実際に提出からシフト作成まで完結するWebアプリの作成です。
しかし、最低限の提出とシフト作成ができるWebアプリをDjangoで実装したのですがやはり保守の面で厳しいものがありました。（主にサーバー代と改良の手間）

そこで最終的には、「調整さん」＋シフト作成のツールという組み合わせで運用を行っています。

「調整さんhttps://chouseisan.com/」さんは実際にそのサイトでどんな感じかみてもらった方が早いと思います。
ざっくばらんに言うと日程調整をWeb上で行うことができ、CSV形式でダウンロードができるウェブサービスです。
つまり提出の面では全て調整さんにお任せすることで

• データベースが不要になる。

• サーバーを必要としない。

と言うメリットがありました。

さて本題のシフトをどうやって自動で作成するのかという話ですが、シフトの自動作成は数学的な問題としても取り扱われ、ナーススケジューリング問題といいます。
興味がある方は以下を参考に。
www.nurse-scheduling-software.com

この問題について組み合わせ最適化を使って問いちゃおうというのが今回の本題ですが、
qiita.com
を参考にして作りました。

以下にその内容を頑張って解読したものがあるので参考までに

コードを書く前の準備

Python 3.7.2
pandas 0.25
numpy 1.17.2
を使ってデータを成形します。

pulp 1.6.10
で離散最適化を行います。

この辺りのライブラリの使い方については後のコード解説の部分で説明します。

前提条件として私のアルバイト先は一般的なナーススケジューリング問題に比べて簡単な条件になっています。

まず1日の時間帯が夜のみです。
一般的には朝、昼、晩であったり、１時間単位でのアサインが想定されているはずです。
そのためとっかかりとしてはいい問題だったかもしれません。

そして今から述べる条件に対して厳密解を求めるのではなくあくまで近似解を求めるという事に気をつけねばなりません。
つまりどの条件に対して厳しくするのかは自分で決めなければなりません。

まず必ず守らなければならない制約が

• 平日は３人、土日祝は５人アサインしなければならない。
• ×と提出された場合はその人をアサインしてはいけない。○と提出した人のみで1日のアサインを決める。

の２つです。

そして守れていないとペナルティーを与える制約が

• バイトメンバーを「たくさん入りたい人」、「普通くらい入りたい人」、「少なめがいい人」の３つに分類してそれぞれの分類で１ヶ月のアサイン数を均等にする。
• アサインしている人のうち、男性よりも女性の方が多くアサインしていないといけない（３人のうち男が２人とかは禁止）。
• 新人のみがアサインしてはいけない（歴の長い人が最低一人アサインしなければならない）。
• ３連勤してはいけない（２連勤はペナルティーなし）。
• アサインの間隔は1週間まで （前回アサインした日から1週間以上空けてはいけない）。

の５つである。
どこのバイト先にもこれくらいの制約はあるように思います。

初めのうちはこのくらいの制約で上手く運用できていたのですが、「（卒論、期末試験、就活、旅行）が忙しくて今月あまり入れません」と言う人が固まって出てくると悲惨なことにペナルティー過多でほとんど機能していないようなシフトが出来上がります。まあこの辺りはプログラムの問題というよりはアルバイトの運営を上手くこなせという感じがしますが。

そこで新たなルールとして固定機能をつけました。

その人が何らかの理由で入れない日が多かったり全体的に人が足りてない時はアサインできる日が少ない人（×ばかりでシフトを提出した人）を「固定」という分類に入れます。
固定に入った人はペナルティの制約を無視して、その月のアサインできる日全てにアサインしてもらうという方法です。
この機能はかなり特殊な状況な気がしますがこういうこともできますよという一例です。

実際はどうやって解いているのか

少し難しくなってきます。注意してお読みください。

前の章でペナルティーの話が出てきましたがこのペナルティーが最小であるシフトは一番いいシフトであると考えられます。
つまりペナルティーの値が０ならそれは制約をきちんと守れた最高のシフトであるということです。

このペナルティーを最小にするという部分について計算してもらうということです。
さて、そのために定式化をしなければなりません。

例えば３連勤をしている人が５人いました。
その場合は「３連勤をした時のペナルティー」×５のようなペナルティーを加算します。

この「３連勤をした時のペナルティー」というのはユーザーが自分で設定しなければなりません。
このように制約に対してペナルティーをどれほど与えるのかというのがまず初めの第一歩です。

これを決めた後はシフトについてどのように表現するかです。
私は今回はシフトに入っている人を1,入らなかった人を0という風に表現しました。

つまりこのような感じです。

０と１ばかりで何が何かさっぱりかもしれません。
列は日付、行は人を表している表と見れば少しわかりやすくなるかもしれません。

横一行で1の数の合計は３もしく5になっていると思います。つまり1日にアサインしている人数です。
縦一列で３連続で1になっているところはほとんどないはずです。つまり３連勤は回避しているということです。

何となくイメージをつかめていただけましたでしょうか。

つまりこの１、０の表を使うことで縦一列の１の合計でその月のアサイン数を見ることができます。
縦一列で連続して１があるなら連勤になってしまっているとわかります。
このようにシフトを0,1で表現することで計算に用いることができ、ペナルティーを計算できるというわけです。

シフトを提出してもらった中で○をつけてくれた人に1,×をつけた人に0という風に表を作成し、その内１の部分を最終的なシフトで０か１に決定していくことでシフトが作られます。（シフト提出で×をつけた人がその日にアサインすることはない、つまり最終的なシフトで必ず０になるはずである）

さてどのように０か１を確かめていくのかは制約で考えられる通りの総当たりです（と勝手に思っている）。
その内ペナルティーが低かったものを採択することで完成です。

実際の定式化については
xは実際に入るかどうかについての変数を表し、0なら入らない、1なら入る。
iは人を表し、jは日を表す。
sは提出してもらうシフトで入れると回答してものは1,入れないには0とする。
$$ x_{ij} ∈ \{0,1\} $$
$$ s_{ij} ∈ \{0,1\} $$
制約１　平日は３人、土日祝は５人アサインしなければならない。
$$ \sum_{i} x_{ij}=3 or 5 $$
制約2 ×と提出された場合はその人をアサインしてはいけない。○と提出した人のみで1日のアサインを決める。
$$ if s_{ij} == 0 x_{ij} = 0 $$
制約１、２に関しては守れない場合はそもそもシフト作成ができないのでこの部分は条件を満たさなければなりません。
以降の制約は制約という形を取っていますが変数を制約に入れることで変数の大きさがペナルティーの大きさに直結します。

制約３　「多め」、「普通」、「少なめ」のそれぞれについてアサイン数の最大、最小を任意に定め、それぞれ超えた分についてペナルティを課す。
$$ Vshiftmax_{j} >= 0,Vshiftmax_{j} >= 0 $$
$$ \sum_{i∈多めの人}\sum_{j} x_{ij} + Vshiftmax_{j} <= many_{max} $$
$$ \sum_{i∈多めの人}\sum_{j} x_{ij} – Vshiftmin_{j} <= many_{min} $$
VshiftmaxとVshiftminは自動的に最小をとるものを選択します。
例えば多めに入りたいというA,B,Cさんがいた時「多め」の最大アサイン数を１０、最小アサイン数を８とした時
Aさんがアサイン数１２、Bさんがアサイン数９、Cさんがアサイン数７という場合が出てきた時
AさんのVshiftmax=2、Vshiftmin=0,
BさんのVshiftmax=０、Vshiftmin=0
CさんのVshiftmax=０、Vshiftmin=１
という風になりVshiftmaxの全員の合計は２、Vshiftminの合計は１
これがそれぞれペナルティーとなります。任意の係数をかけて制約ごとに重要度を決めます。
制約４　全ての日において新人のみがアサインしないようにする。
$$ Vshiftnew_{j} >= 0$$
\sum_{j}\sum_{i∈ベテランの人} x_{ij} + Vshiftnew_{j} <= 1
ベテランが一人もいない場合Vshiftnewは１になりペナルティーに計上され、ベテランが一人でもいればVshiftnewは０となりペナルティーはなしになります。
制約５　３連勤してはいけない（２連勤はペナルティーなし）。
制約６　アサインの間隔は1週間まで （前回アサインした日から1週間以上空けてはいけない）。
それぞれ式は省略するが制約３、４と同様に制約の中にV〜という変数を含めその値が０になればペナルティーはなしという仕組みです。

V〜を含めた目的関数が最小となるようにPythonが計算をグルグル回してくれるのでここが自動という所以でしょうか。
コードにおいての目的関数決定の部分です。

ざっくりとわかってもらえたでしょうか。文章を書くのがとても苦手なので伝わりずらかった部分もあるかもしれませんが、随時わかりやすくなるように更新していく所存であります。

コード解説

import numpy as np, pandas as pd
from pulp import *
from ortoolpy import addvars, addbinvars
import random
import datetime
import openpyxl
'''
Vは変数を表している
Cは定数を表している
shift.index は[1,,,,,30]
shift.columns は　[Aさん、・・・・Zさん]
1が入れる希望シフトでなおかつ最終的に入る人を示す
０はシフト禁止でなおかつ最終的に入らない人を示す。
'''
# ペナルティの重み定数
C_need_dif = 1000
C_lock = 50
C_3continuous_work = 20
C_2continuous_work = 3
C_average = 10
C_gender = 3
C_experience = 3
C_blank = 1
# ここでExcelのデータを取得
shift = pd.read_excel('1月作成用.xlsx',sheet_name='Shift',index_col=0)
manage = pd.read_excel('1月作成用.xlsx',sheet_name='Manage',index_col=0)
member = pd.read_excel('1月作成用.xlsx',sheet_name='Member',index_col=0)
member = member.T
detail = pd.read_excel('1月作成用.xlsx',sheet_name='Detail',index_col=0)
# Shiftシートのカラムとインデックス
date = shift.index
employee = shift.columns
# 確認用にコピー
shift_show = shift
# 従業員数と月の日数
days, number_employee = shift.shape[0], shift.shape[1]
holidays = play_ground.holiday_list(days)
# 変数作成
var = pd.DataFrame(np.array(addbinvars(days, number_employee)), columns=employee, index=date)
# ０と１を逆転させている
shift_rev = shift[shift.columns].apply(lambda r: 1-r[shift.columns],1)
k = LpProblem()
# 希望していない場所には入らないようにする。
for (_, h),(_, n) in zip(shift_rev.iterrows(),var.iterrows()):
k += lpDot(h, n) <= 0
# 変数の追加
shift['V_need_dif'] = addvars(days)         # 必要人数に達していない時のペナルティ変数
shift['V_gender_rate'] = addvars(days)      # 女性が少ない時のペナルティ
shift['V_experience'] = addvars(days)       # 新人だけになった時のペナルティ
shift['need'] = manage.need                 # 必要人数
V_3continuous_work = np.array(addbinvars(days-2, number_employee))
V_2continuous_work = np.array(addbinvars(days-1, number_employee))
V_blank = np.array(addbinvars(days-6, number_employee))
V_max = addvars(number_employee)
V_min = addvars(number_employee)
V_lock = addvars(number_employee)
# 足りない人が入れば終了
shortage = []
for index,r in shift[employee].iterrows():
if sum(r) < int(shift.at[index,'need']):
shortage.append(index)
if shortage:
day = ''
for i in shortage:
day += (str(i) + '日足りません')
print(day)
sys.exit()
# 必要な人数に対するペナルティーを求める。
for (_, r),(_, d) in zip(shift.iterrows(),var.iterrows()):
k += r.V_need_dif >= (lpSum(d) - r.need)
k += r.V_need_dif >= -(lpSum(d) - r.need)
# 連勤に対してペナルティーを求める。
for i in list(range(number_employee)):
for n,p in enumerate((var.values[:-2,i] + var.values[1:-1,i] + var.values[2:,i]).flat):
k += p - V_3continuous_work[n][i] <= 2
for i in list(range(number_employee)):
for n,p in enumerate((var.values[:-1,i] + var.values[1:,i]).flat):
k += p - V_2continuous_work[n][i] <= 1
# 長く空きすぎるとペナルティ
for i in list(range(number_employee)):
for n,p in enumerate((var.values[:-6,i] + var.values[1:-5,i] + var.values[2:-4,i] + var.values[3:-3,i]+ var.values[4:-2,i]+ var.values[5:-1,i]+ var.values[6:,i]).flat):
k += p + V_blank[n][i] >= 1
# ある程度の入る量を調整
amount_more_user_list = member[member['amount'].isin([0])].index
amount_normal_user_list = member[member['amount'].isin([1])].index
amount_less_user_list = member[member['amount'].isin([2])].index
amount_zero_user_list = member[member['amount'].isin([3])].index
amount_lock_user_list = member[member['amount'].isin([4])].index
# 入りすぎ、入らなすぎにペナルティを与える変数
V_shift_max = pd.DataFrame(V_max,index=employee).T
V_shift_min = pd.DataFrame(V_min,index=employee).T
V_shift_lock = pd.DataFrame(V_lock,index=employee).T
# Excelから取得したデータをもとに
amount_more_setting = {'max': detail.at['max','more'], 'min': detail.at['min','more']}
amount_normal_setting = {'max': detail.at['max','normal'], 'min': detail.at['min','normal']}
amount_less_setting = {'max': detail.at['max','less'], 'min': detail.at['min','less']}
# 制約
for name, r in var[amount_more_user_list].iteritems():
k += lpSum(r) + V_shift_max.at[0,name] >= amount_more_setting['min']
k += lpSum(r) - V_shift_min.at[0,name] <= amount_more_setting['max']
for name, r in var[amount_normal_user_list].iteritems():
k += lpSum(r) + V_shift_max.at[0,name] >= amount_normal_setting['min']
k += lpSum(r) - V_shift_min.at[0,name] <= amount_normal_setting['max']
for name, r in var[amount_less_user_list].iteritems():
k += lpSum(r) + V_shift_max.at[0,name] >= amount_less_setting['min']
k += lpSum(r) - V_shift_min.at[0,name] <= amount_less_setting['max']
for (_, h),(name, n) in zip(shift[amount_lock_user_list].iteritems(),var[amount_lock_user_list].iteritems()):
k += lpSum(n) + V_shift_lock.at[0,name] >= lpSum(h)
k += lpSum(n) - V_shift_lock.at[0,name] <= lpSum(h)
# 男女偏り、新人のみにならない
woman_list = member[member['sex'].isin([1])].index
veteran_list = member[member['rank'].isin([1])].index
for (_,r),(_,d) in zip(shift.iterrows(),var[woman_list].iterrows()):
if r.need == 0:
pass
else:
k += (r.V_gender_rate + lpSum(d)) >= 2
for (_,r),(_,d) in zip(shift.iterrows(),var[veteran_list].iterrows()):
if r.need == 0:
pass
else:
k += (r.V_experience + lpSum(d)) >= 1
# 目的関数決定
k += C_need_dif * lpSum(shift.V_need_dif) \
+ C_3continuous_work * lpSum(V_3continuous_work) \
+ C_2continuous_work * lpSum(V_2continuous_work) \
+ C_average * lpSum(V_max) \
+ C_average * lpSum(V_min) \
+ C_gender * lpSum(shift.V_gender_rate) \
+ C_experience * lpSum(shift.V_experience) \
+ C_lock * lpSum(V_lock) \
+ C_blank * lpSum(V_blank)
k.solve()
result = np.vectorize(value)(var).astype(int)
R_continuous_work = np.vectorize(value)(V_2continuous_work).astype(int)
print('目的関数', value(k.objective))
print(result)
continuous_work_list = []
for i in np.sum(R_continuous_work,axis=0):
if i >= 1:
continuous_work_list.append('有')
else:
continuous_work_list.append('無')
fi = []
for cou,r in enumerate(result):
fi.append([])
for i,j in zip(r,shift.columns):
if i*j != '':
fi[cou].append(i*j)
amount = []
member_rev = member.T
for name, r in member_rev.iteritems():
if r.amount == 0:
amount.append("多め")
elif r.amount == 1:
amount.append("普通")
elif r.amount == 2:
amount.append("少なめ")
elif r.amount == 3:
amount.append("無し")
elif r.amount == 4:
amount.append("固定")
count = np.sum(result, axis=0)
evaluation = pd.DataFrame([amount,count,continuous_work_list],
index=['シフト希望量', '入る量', '連勤'],columns=employee)
print(evaluation)
# ここで目視用のシートを作る
result_show_color = pd.DataFrame(result,index=date,columns=employee)
for index,i in result_show_color.iterrows():
for column, c in i.iteritems():
if c == 1: # 実際入ってる人
pass
elif c == 0 and shift.at[index,column] == 1: # 入る希望はあったが入らない人
pass
else:
result_show_color.at[index,column] = 2 # 入るつもりがない人
shift_member = fi
for values in shift_member:
random.shuffle(values)
shift_result = pd.DataFrame(shift_member,index=date)
with pd.ExcelWriter('1月.xlsx') as writer:
shift_result.to_excel(writer, sheet_name='Result')
evaluation.to_excel(writer, sheet_name='Evaluation')
result_show_color.to_excel(writer,sheet_name="show")

長ったらしいコードが続いてしまったがやっていることは前章で述べた通りです。
pandasの基本を抑えていればそこまで解読も大変でないはず。。。
ただ人に見られる事を想定したコードではないので多少見苦しい部分はあると思います。
読みにくかったらすみません。

コメントで分からないところがあれば解説します。

振り返り

基本２、3秒で計算が終わるので大変上手く行ったと個人的には思っています。
しかし結局のところ人が足りなければ何の意味もないので最終的に重要になるのは現実での運営みたいなところがあるかもしれません。
身もふたもない話ですが。