diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..ca8eaac --- /dev/null +++ b/.coveragerc @@ -0,0 +1,8 @@ +[run] +omit = + ./tests/* + ./venv/* + ./docs/* +[report] +precision = 2 +show_missing = True \ No newline at end of file diff --git a/.gitignore b/.gitignore index b6e4761..288c2cf 100644 --- a/.gitignore +++ b/.gitignore @@ -127,3 +127,6 @@ dmypy.json # Pyre type checker .pyre/ + +#Pycharm +.idea diff --git a/README.md b/README.md index c547390..8aab049 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,27 @@ # pyworkforce -Common tools for workforce management and production schedule and optimization +Common tools for workforce management, schedule and optimization problems. + + +### Queue systems: + +It can be used for solving the required number of positions to manage a number of transactions, +under some systems pre-defined parameters and goals. + + +#### Example: + +```python +from pyworkforce.queuing.erlang import ErlangC + +erlang = ErlangC(transactions=100, asa=20/60, aht=3, interval=30, max_occupancy=0.85, shrinkage=0.3) + +positions_requirements = erlang.required_positions(service_level=0.8) +print("positions_requirements: ", positions_requirements) + + +>> positions_requirements: {'raw_positions': 14, + 'positions': 20, + 'service_level': 0.8883500191794669, + 'occupancy': 0.7142857142857143, + 'waiting_probability': 0.1741319335950498} +``` diff --git a/dev-requirements.txt b/dev-requirements.txt new file mode 100644 index 0000000..32cb212 --- /dev/null +++ b/dev-requirements.txt @@ -0,0 +1,4 @@ +pytest==6.2.2 +codecov==2.1.11 +pytest-cov==2.11.1 +twine==2.4.7 \ No newline at end of file diff --git a/examples/queing/erlangc.py b/examples/queing/erlangc.py new file mode 100644 index 0000000..f171b6e --- /dev/null +++ b/examples/queing/erlangc.py @@ -0,0 +1,30 @@ +""" +Example taken from: https://www.callcentrehelper.com/erlang-c-formula-example-121281.htm + +Requirement: Find the number of agents required to manage call center transactions +under the following parameters: + +Number of calls: 100 +In a period of minutes: 30 +Average Handling Time (seconds): 180 (3 minutes) +Required Service Level: 80% +Target Answer Time (Seconds): 20 +Maximum Occupancy: 85% +Shrinkage: 30% +""" + +from pyworkforce.queuing.erlang import ErlangC + +erlang = ErlangC(transactions=100, asa=20/60, aht=3, interval=30, max_occupancy=0.85, shrinkage=0.3) + +positions_requirements = erlang.required_positions(service_level=0.8) +print("positions_requirements: ", positions_requirements) + +achieved_service_level = erlang.service_level(positions=positions_requirements['raw_positions']) +print("achieved_service_level: ", achieved_service_level) + +waiting_probability = erlang.waiting_probability(positions=positions_requirements['raw_positions']) +print("waiting_probability: ", waiting_probability) + +achieved_occupancy = erlang.achieved_occupancy(positions=positions_requirements['raw_positions']) +print("achieved_occupancy: ", achieved_occupancy) diff --git a/pyworkforce/__init__.py b/pyworkforce/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pyworkforce/queuing/erlang.py b/pyworkforce/queuing/erlang.py new file mode 100644 index 0000000..02981e0 --- /dev/null +++ b/pyworkforce/queuing/erlang.py @@ -0,0 +1,98 @@ +from math import exp, ceil, floor + + +class ErlangC: + def __init__(self, transactions: float, asa: float, aht: float, + interval: int = None, shrinkage=0.0, max_occupancy=1.0, + **kwargs): + """ + Computes the number of positions required fo attend a number of transactions in a queue system based on ErlangC + Implementation based on: https://lucidmanager.org/data-science/call-centre-workforce-planning-erlang-c-in-r/ + + :param transactions: number of total transactions that comes in + :param aht: average handling time of a transaction (minutes) + :param interval: Interval length (minutes) + :param shrinkage: Percentage of time that an operator unit is not available + :param max_occupancy: Maximum percentage of time that an attending position can be occupied + """ + self.n_transactions = transactions + self.asa = asa + self.aht = aht + self.interval = interval + self.intensity = (self.n_transactions / self.interval) * self.aht + self.shrinkage = shrinkage + self.max_occupancy = max_occupancy + + def waiting_probability(self, positions, scale_positions=False): + """ + :param positions: Number of positions to attend the transactions + :param scale_positions: True if the positions where calculated using shrinkage + :return: the probability of a transaction waits in queue + """ + if scale_positions: + productive_positions = floor((1 - self.shrinkage) * positions) + else: + productive_positions = positions + + erlang_b_inverse = 1 + for position in range(1, productive_positions + 1): + erlang_b_inverse = 1 + (erlang_b_inverse * position / self.intensity) + + erlang_b = 1 / erlang_b_inverse + return productive_positions * erlang_b / (productive_positions - self.intensity * (1 - erlang_b)) + + def service_level(self, positions, scale_positions=False): + """ + :param positions: Number of positions attending + :param scale_positions: True if the positions where calculated using shrinkage + :return: achieved service level + """ + if scale_positions: + productive_positions = floor((1 - self.shrinkage) * positions) + else: + productive_positions = positions + + probability_wait = self.waiting_probability(productive_positions, scale_positions=False) + exponential = exp(-(productive_positions - self.intensity) * (self.asa / self.aht)) + return max(0, 1 - (probability_wait * exponential)) + + def achieved_occupancy(self, positions, scale_positions=False): + """ + :param positions: Number of raw positions + :param scale_positions: True if the positions where calculated using shrinkage + :return: Expected occupancy of positions + """ + if scale_positions: + productive_positions = floor((1 - self.shrinkage) * positions) + else: + productive_positions = positions + + return self.intensity/productive_positions + + def required_positions(self, service_level): + """ + :param service_level: Target service level + :return: Number of positions needed to ensure the required service level + """ + positions = round(self.intensity + 1) + achieved_service_level = self.service_level(positions, scale_positions=False) + while achieved_service_level < service_level: + positions += 1 + achieved_service_level = self.service_level(positions, scale_positions=False) + + achieved_occupancy = self.achieved_occupancy(positions, scale_positions=False) + + if achieved_occupancy > self.max_occupancy: + raw_positions = ceil(self.intensity / self.max_occupancy) + achieved_occupancy = self.achieved_occupancy(raw_positions) + achieved_service_level = self.service_level(raw_positions) + + raw_positions = ceil(positions) + waiting_probability = self.waiting_probability(positions=raw_positions) + positions = ceil(raw_positions / (1 - self.shrinkage)) + + return {"raw_positions": raw_positions, + "positions": positions, + "service_level": achieved_service_level, + "occupancy": achieved_occupancy, + "waiting_probability": waiting_probability} diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..25dac4e --- /dev/null +++ b/setup.py @@ -0,0 +1,28 @@ +import pathlib +from setuptools import setup + +# The directory containing this file +HERE = pathlib.Path(__file__).parent + +# The text of the README file +README = (HERE / "README.md").read_text() + +# This call to setup() does all the work +setup( + name="pyworkforce", + version="0.1.0", + description="Common tools for workforce management, schedule and optimization problems", + long_description=README, + long_description_content_type="text/markdown", + url="https://github.com/rodrigo-arenas/pyworkforce", + author="Rodrigo Arenas", + author_email="rodrigo.arenas456@gmail.com", + license="MIT", + classifiers=[ + "License :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + ], + packages=["pyworkforce"], + include_package_data=True, +) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_queing.py b/tests/test_queing.py new file mode 100644 index 0000000..e8421c0 --- /dev/null +++ b/tests/test_queing.py @@ -0,0 +1,16 @@ +from pyworkforce.queuing.erlang import ErlangC + + +def test_expected_erlangc_results(): + erlang = ErlangC(transactions=100, asa=0.33, aht=3, interval=30, shrinkage=0.3) + results = erlang.required_positions(service_level=0.8) + raw_positions = results['raw_positions'] + positions = results['positions'] + service_level = results['service_level'] + occupancy = results['occupancy'] + + assert raw_positions == 14 + assert positions == 20 + assert round(service_level, 3) == 0.888 + assert round(occupancy, 3) == 0.714 +