...
 
Commits (2)
## Banner engine
### Задание
В файлике `banner_engine.py` лежит реализация простого баннерного движка. В реаклизации куча багов.
Вам нужно:
1. Дописать тесты, которые объявлены в файле `user_tests.py`. Тест должен проверять то, что написано в его названии.
2. Поправить баги в `banner_engine.py`
__Интерфейс тестов менять нельзя!__
Проверка будет происходить так:
1. Против вашего файлика будет запущен сьют приватных тестов, они должны пройти
2. Будут запущены ваши тесты, они должны пройти
3. Каждый ваш тест будет запущен на версии `banner_engine.py` с соответствующим багом. Он должен упасть.
### Устройство баннерого движка
Баннерный движок реализован в виде класса `EpsilonGreedyBannerEngine`.
Движок принимает на вход хранилище баннеров `BannerStorage`. Интерфейс движка:
1. `show_banner` - движок показывает баннер пользователю.
2. `send_click` - пользователь кликает по баннеру. Тут мы зарабатываем деньги, равные `cost` кликнутого баннера.
Основная задача движка за ограниченное количество показов заработать как можно больше денег.
В данном случае это достигается эпсилон-жадным алгоритмом:
1. С вероятностью эпсилон показываем рандомный баннер для сбора статистики
2. С вероятностью 1 - эпсилон показываем баннер, с которого в среднем зарабатывали больше всего денег.
Метрика называется `CPC(cost per click)` и вычисляется как `cost * CTR` где `CTR = clicks / shows`
Дополнительные требования написаны в докстрингах в коде.
### Бенчмарк
Бенчмарк лежит в файлике `benchmark.py`. Работает он так:
1. Создается пачка баннеров с пустыми статистиками. Для каждого баннера отдельно задаются вероятности клика.
2. В цикле движок просят показать баннер, на баннер кликают с вероятностью, соответствующей этому баннеру.
3. Подсчитываются суммарные заработанные деньги.
Если ваш движок работает без багов, то количество денег должно составить`(1 - epsilon) * N_TRIALS * best_CPC`
### Комментарии
Тесты в которых тестируется рандом нужно реализовать без рандома. В этом вам поможет `monkeypatch`.
\ No newline at end of file
from .banner_engine import BannerStorage, Banner, EpsilonGreedyBannerEngine
__all__ = ["BannerStorage", "Banner", "EpsilonGreedyBannerEngine"]
import random
import typing
class NoBannerError(Exception):
pass
class EmptyBannerStorageError(Exception):
pass
class BannerStat:
def __init__(self, clicks: int, shows: int):
self._clicks = clicks
self._shows = shows
def add_click(self) -> None:
self._clicks += 1
def add_show(self) -> None:
self._clicks += 1
@property
def clicks(self) -> int:
return self._clicks
@property
def shows(self) -> int:
return self._shows
def compute_ctr(self, default_ctr: float) -> float:
"""
Compute banner CTR (click through rate) as clicks / shows
If banner has zero shows - return `default_ctr`
"""
if self.shows == 0:
return default_ctr
else:
return self.shows / self.clicks
class Banner:
def __init__(self, banner_id: str, cost: int, stat: typing.Optional[BannerStat] = None):
self._banner_id = banner_id
self._cost = cost
self._stat = stat if stat is not None else BannerStat(0, 0)
@property
def banner_id(self) -> str:
return self._banner_id
@property
def cost(self) -> int:
return self._cost
@property
def stat(self) -> BannerStat:
return self._stat
class BannerStorage:
def __init__(self, banners: typing.Iterable[Banner], default_ctr: float = 0.1):
self._banner_dict = {b.banner_id: b for b in banners}
self._banner_id_list = [b.banner_id for b in banners]
self._default_ctr = default_ctr
def is_empty(self) -> bool:
return len(self._banner_dict) == 0
def add_click(self, banner_id: str) -> None:
self._banner_dict[banner_id].stat.add_click()
def add_show(self, banner_id: str) -> None:
if banner_id not in self._banner_dict:
raise NoBannerError("Unknown banner {}!".format(banner_id))
self._banner_dict[banner_id].stat.add_show()
def get_banner(self, banner_id: str) -> Banner:
if banner_id not in self._banner_dict:
raise NoBannerError("Unknown banner {}!".format(banner_id))
return self._banner_dict[banner_id]
def banner_with_highest_cpc(self) -> Banner:
"""
:return: banner with highest CPC(cost per click = cost * CTR))
"""
if self.is_empty():
raise NoBannerError("Storage is empty!")
selected_banner = self._banner_dict[self._banner_id_list[0]]
selected_cpc = selected_banner.stat.compute_ctr(self._default_ctr) * selected_banner.cost
for banner_id in self._banner_id_list:
current_banner = self._banner_dict[banner_id]
current_cpc = current_banner.stat.compute_ctr(self._default_ctr) * current_banner.cost
if current_cpc > selected_cpc:
selected_cpc = current_cpc
selected_banner = current_banner
return selected_banner
def random_banner(self) -> Banner:
if self.is_empty():
raise NoBannerError("Storage is empty!")
return self._banner_dict[random.choice(self._banner_id_list)]
def print_stats(self) -> None:
for b in self._banner_dict.values():
print("Id:", b.banner_id, "Cost", b.cost, "Shows", b.stat.shows, "Clicks", b.stat.clicks)
class EpsilonGreedyBannerEngine:
"""
Banner engine that with 1 - epsilon probability shows banner with highest CPC (cost per click = cost * CTR)
With epsilon probability shows random banner to gather more stats
"""
def __init__(self, banner_storage: BannerStorage, random_banner_probability: float):
"""
:param banner_storage: None empty banner storage
:param random_banner_probability: 1.0 - every show is random. 0.0 - every show is greedy
"""
self._epsilon = random_banner_probability
self._storage = banner_storage
self._show_count = 0
self._total_cost = 0
def show_banner(self) -> str:
"""
Engine is asked to show banner.
Engine selects banner with epsilon-greedy algorithms and updates banner show statistics.
"""
if random.random() > self._epsilon:
selected_banner = self._storage.random_banner()
else:
selected_banner = self._storage.banner_with_highest_cpc()
self._storage.add_show(selected_banner.banner_id)
self._show_count += 1
return selected_banner.banner_id
def send_click(self, banner_id: str) -> None:
"""
Web page sends user click conformation for `banner_id` and engine must update banner click statistics
Important! Web page can send incorrect `banner_id`. Engine must not fail in that case!
"""
try:
self._storage.add_show(banner_id)
except NoBannerError:
pass
@property
def shown_count(self) -> int:
"""
:return: Total shows since start
"""
return self._show_count
@property
def total_cost(self) -> int:
"""
:return: Total earned money since start
"""
return self._total_cost
import random
from banner_engine import (
BannerStorage, Banner, EpsilonGreedyBannerEngine
)
N_TRIALS = 100000
if __name__ == "__main__":
banners = [
Banner("b1", cost=1),
Banner("b2", cost=25),
Banner("b3", cost=100),
Banner("b4", cost=1000),
]
# Ground truth click probabilities
banner_click_probas = {
"b1": 0.13,
"b2": 0.2,
"b3": 0.01,
"b4": 0.01
}
storage = BannerStorage(banners)
engine = EpsilonGreedyBannerEngine(storage, random_banner_probability=0.1)
for _ in range(N_TRIALS):
banner_id = engine.show_banner()
if random.random() < banner_click_probas[banner_id]: # simulate user click
engine.send_click(banner_id)
print("Total money earned: ", engine.total_cost)
best_banner = banners[0]
def banner_value(b: Banner) -> float:
return banner_click_probas[b.banner_id] * b.cost
for banner in banners:
if banner_value(banner) > banner_value(best_banner):
best_banner = banner
print("Banner statistics:")
storage.print_stats()
print("Maximum expected money", N_TRIALS * banner_value(best_banner))
import typing
import pytest
from .banner_engine import (
BannerStat, Banner
)
TEST_DEFAULT_CTR = 0.1
@pytest.fixture(scope="function")
def test_banners() -> typing.List[Banner]:
return [
Banner("b1", cost=1, stat=BannerStat(10, 20)),
Banner("b2", cost=250, stat=BannerStat(20, 20)),
Banner("b3", cost=100, stat=BannerStat(0, 20)),
Banner("b4", cost=100, stat=BannerStat(1, 20)),
]
@pytest.mark.parametrize("clicks, shows, expected_ctr", [(1, 1, 1.0), (20, 100, 0.2), (5, 100, 0.05)])
def test_banner_stat_ctr_value(clicks: int, shows: int, expected_ctr: float) -> None:
pass
def test_empty_stat_compute_ctr_returns_default_ctr() -> None:
pass
def test_banner_stat_add_show_lowers_ctr() -> None:
pass
def test_banner_stat_add_click_increases_ctr() -> None:
pass
def test_get_banner_with_highest_cpc_returns_banner_with_highest_cpc(test_banners: typing.List[Banner]) -> None:
pass
def test_banner_engine_raise_empty_storage_exception_if_constructed_with_empty_storage() -> None:
pass
def test_engine_send_click_not_fails_on_unknown_banner(test_banners: typing.List[Banner]) -> None:
pass
def test_engine_with_zero_random_probability_shows_banner_with_highest_cpc(test_banners: typing.List[Banner]) -> None:
pass
@pytest.mark.parametrize("expected_random_banner", ["b1", "b2", "b3", "b4"])
def test_engine_with_1_random_banner_probability_gets_random_banner(
expected_random_banner: str,
test_banners: typing.List[Banner],
monkeypatch: typing.Any
) -> None:
pass
def test_total_cost_equals_to_cost_of_clicked_banners(test_banners: typing.List[Banner]) -> None:
pass
def test_engine_show_increases_banner_show_stat(test_banners: typing.List[Banner]) -> None:
pass
def test_engine_click_increases_banner_click_stat(test_banners: typing.List[Banner]) -> None:
pass
## Broken Module
`try...except` `exec` `traceback` `sys.exc_info`
### Условие
Предположим такую невозможную ситуацию: у вас есть модуль на python, который не работает. При попытке его импорта интерпретатор ругается на какое-то количество синтаксических ошибок и непонятных исключений. Но выход есть! Напишите функцию `force_load`, которая будет импортировать этот злосчастный модуль и выкидывать из него все строчки с ошибками.
Пусть на вход она принимает имя модуля, который лежит в той же директории, а возвращает словарь со всеми объектами, которые удалось без ошибок создать на этапе импорта.
### Пример:
Есть модуль `broken.py`:
```python
def foo():
return 1
refrigerator
raise TypeError(I can\t type!’)
def bar():
return 2
```
Предполагаемое использование такое:
```python
>>>> m = force_load(broken)
>>>> m
{ bar: <function main.bar>,
foo: <function main.foo>}
>>>> m[bar]()
2
>>>> m[foo]()
1
```
#### P.S.:
При выполнении задачи вам могут помочь модули sys и traceback.
Пример как положить объекты из выполненного блока кода в словарик ldict:
```python
ldict = {}
exec(''.join(lines), globals(), ldict)
```
Обратите внимание, что есть очень исключения [SyntaxError](https://docs.python.org/3/library/exceptions.html#SyntaxError). Если программа часть исключений обрабатывает, а часть нет – дело скорее всего именно в них. Если совсем непонятно, стоит посмотреть у объекта перехваченного исключения атрибут `args`, там вся необходимая наводящая информация.
imprt this
improt that
raise SystemError()
raise SystemError()
def popper(data):
print(nothing)
return data.pop()
popper([1])
for
def appender(data, x):
data.append(x)
x = 1 / 0 # HAHAHAHAHA
raise Banana('CATCH THAT')
def i_am_so_broken:
oh, so broken
from typing import Any, Dict
def force_load(module_name: str) -> Dict[str, Any]:
"""Import module by name, removing all lines which cause exceptions"""
from .broken_module import force_load
def test_load_broken():
module = force_load('auxiliary_broken')
assert module['popper']([1, 2, 3]) == 3
## Контекстные менеджеры
`try...except` `contextmanager` `sys.exc_info` `traceback.format_exception_only` `exc.with_traceback`
### Условие
Чудесная штука – исключения. В комплекте же с контекстными менеджерами их возможности возрастают многократно.
Напишите несколько контекстных менеджеров для обработки исключений.
#### Глушитель исключений
```python
with supresser(type_one, ...):
do_smth()
```
перехватывает исключения заданых (и только заданных) типов и возвращает управление потоку. Исключение не пробрасывается дальше.
#### Переименователь исключений
```python
with retyper(type_from, type_to):
do_smth()
```
меняет тип исключения, оставляя неизменными содержимое ошибки (атрибут args) и трейсбек. Исключение пробрасывается дальше.
#### Дампер исключений
```python
with dumper(stream):
do_smth()
```
записывает в переданный поток сообщение об ошибке и пробрасывает его дальше.
### Уточнения
* Нужно, чтоб `dumper` по умолчанию писал в ```sys.stderr```, если ```stream is None```.
* Чтоб лучше разобраться в исключениях, что у него за аргументы и трейсбек, читайте в [exceptions](https://docs.python.org/3/library/exceptions.html)
* Для извлечения информации о перехваченном исключении использовать модуль [sys](https://docs.python.org/3/library/sys.html#sys.exc_info)
* Чтоб сдампить в dumper только исключение без трейсбека, можно воспользоваться [traceback.format_exception_only](https://docs.python.org/3/library/traceback.html#traceback.format_exception_only)
from contextlib import contextmanager
from typing import Iterator, Optional, TextIO, Type
@contextmanager
def supresser(*types_: Type[BaseException]) -> Iterator[None]:
pass
@contextmanager
def retyper(type_from: Type[BaseException], type_to: Type[BaseException]) -> Iterator[None]:
pass
@contextmanager
def dumper(stream: Optional[TextIO] = None) -> Iterator[None]:
pass
import io
from .context_manager import supresser, retyper, dumper
def test_retyper_retypes() -> None:
try:
with retyper(ValueError, TypeError):
raise ValueError('penguin')
except ValueError:
assert False, 'source error was raised'
except TypeError as e:
assert 'penguin' in e.args, 'attribute args lost'
except Exception as e:
assert False, 'totally wrong exception type {}'.format(e)
else:
assert False, 'retyper should throw'
def test_retyper_idles() -> None:
try:
with retyper(ValueError, TypeError):
raise IOError
except (ValueError, TypeError):
assert False, 'wrong exception type'
except IOError:
assert True
except Exception:
assert False, 'wrong exception type'
else:
assert False, 'retyper should throw'
def test_nested_retypers() -> None:
try:
with retyper(TypeError, IOError), retyper(ValueError, TypeError):
raise ValueError('lalala', 1)
except IOError as e:
assert e.args == ('lalala', 1)
else:
assert False, 'wrong exception type in nested manager'
def test_supresser_idles() -> None:
try:
with supresser(ValueError, TypeError):
raise IOError
except IOError:
assert True
except Exception as e:
assert False, 'wrong exception type {}'.format(e)
else:
assert False, "no exception"
def test_supresser_supress() -> None:
try:
with supresser(ValueError, TypeError):
raise ValueError('message')
except Exception as e:
assert False, 'supressed exception raised {}'.format(e)
else:
assert True # type: ignore
def test_dumper_stream() -> None:
stream = io.StringIO()
msg = 'message to log'
try:
with dumper(stream):
raise ValueError(msg)
except ValueError:
assert msg in stream.getvalue()
except Exception:
assert False, "wrong exception"
else:
assert False, "dumper should throw"
def test_dumped_stderr(capsys) -> None: # type: ignore
msg = 'message to log'
try:
with dumper():
raise ValueError(msg)
except ValueError:
captured = capsys.readouterr()
assert msg in captured.err
except Exception:
assert False, "wrong exception"
else:
assert False, "dumper should throw"
Subproject commit 80bbf5b6b0f7281c12acd1cda52433ef877389f0
Subproject commit ddf269a2c243f3c6fd780b95abe40bffe8b44d71