List of exercises

1. Stock investment – Complexity analysis, iterators, multiprocessing
2. Money transfer simulator – Package and project lifecycle
3. Bread-First Search in a graph – Time optimization
4. Code breaker – Time optimization
5. Chess master – Asynchronous programming
6. Virus spread simulator – Magic methods, operator overload, iterators, protocols
7. The TODO app API – FastAPI asynchronous endpoints and Pydantic models

Exercise 1: Stock investment

Time complexity analysis and multiprocessing

You are now a wealth advisor for your clients. They require your advice to invest in the most profitable stock portfolio according to their maximum budget.

Available stock come from a data file of 26 stocks that you can propose to clients.

Each stock has:

• a name from "A" to "Z"
• a cost
• a profit after 2 years, %age of the initial cost

Example of portfolio : 3 stocks (E, S, J) for a cost of 60+24+34 and a profit after 2 years of 60*17% + 24*21% + 34*27%

Part I: Implement the business logic

1.1. Download the code .py stub and .json file of available stocks: place them at the root of your Pycharm project.

1.2. Implement get_best_portfolio(...) according to its docstring

1.3. Implement a few tests with some porfolios p, k, l:

1.4. Run a type checker to check that you are not messing up with data types:

1.5. Implement get_best_portfolio_for_budget(...) according to its docstring

1.6. Implement a few tests, reusing the previous test instances, as follows.
stocks needs to be successfully read from the JSON data file.

Messing up with the data inputs and outputs? Get help the type checker! Use mypy or activate the type checker in your IDE.

Before continuing, follow the "complexity, iterators & generators" course

Part II: Plug the business logic to combinatoric data

2.1. Temporarily reduce the input JSON data to 20 stocks instead of 26 in order to reduce compute time. Crop the dataset after stock name T in stocks.json.

2.2. Create a combination_generator generator that yields all 2**20-1 possible combinations of stocks.

• That generator can rely on a combinatoric iterator from itertools
• Number matter: 2 stocks might be more profitable than 3 other ones. You must seek profit for portfolios of any size from 1 stock to 20 stocks

2.3. In main, pass an instanciated generator to get_best_portfolio_for_budget

2.4. Run your code without multiprocessing to get the result.
Expected result: Best is earning 9908 for 498 spent

Part III: Complexity and heuristics

4.1. Evaluate the time and space complexity of your code

This exercise implements the Knapsack problem:

Computation of all results gives optimal results but is way too slow. Most business applications would prefer faster suboptimal results such as heuristics.

4.2 Optional: Implement an heuristics to find a suboptimal solution faster. For instance the greedy algorithm by pre-sorting data on the profit/cost ratio.

Before continuing, follow the "multiprocessing and IPC" course

Part IV: Multiprocessing and IPC

4.1. Implement work(...). This function gets data from the queue which can be:

• an Iterable[Iterable[str]]e.g. [("A", "C"), ("K"), ("L", "B", "F")]
• None if the process must stop because there is no more data

4.2. In __main__, create a process Manager and a Queue to exchange data. Put the jobs to the queue and then a None for each worker.

4.3. In __main__, create a Pool(4) for 4 workers and map or startmap the processes to the data from the queue.

4.4. Once all processes eventually closed, get the final best result before exiting

5.4. Restore all 26 stocks in the JSON file and test.

Expected result: Best is earning 10146 for 499 spent (~ 1'30 to 4' for 1 to 4 CPUs)

Before continuing, follow the "JIT compiling" course

Part V: Use JIT to accelerate computations

With pypy

5.1. Install pypy : sudo apt install pypy3

Since pypy does not implement the very last Python version, you may have to update unsupported syntax. Also some libs or functions may be not supported.

5.2. Runs your single/multiprocess code with pypy, and measure execution time

With numba (works only with numpy arrays)

5.3. From get_best_portfolio_for_budget, extract the most CPU-bound computations into calculate_cost_and_profit(). Modify the prototype to use the njit decorator and numpy.array to encode the portfolio

5.4. Update get_best_portfolio_for_budget so that it converts original portfolio, costs, and profits into numpy.array That conversion has a time cost

5.5. Compare execution time with and without JIT.

Expected result: You should observe an execution time of 3x to 4x with the JIT

Exercise 2. Money transfer simulator

In this exercise we are going to create a simplified Information System that is able to handle and simulate bank transactions.

In our scenario there are 4 actors: a bank (HSBC), a supermarket (Walmart), and 2 individuals Alice and Bob.

Each actor has his/her own bank account.

Part 1: The basic scenario

• 1.1. Create a class BankAccount that owns 2 attributes:

  • a private owner (of type str): the owner's name
  • a protected balance (of type int): the balance (Decimals do not matter)
  • the class constructor takes in parameter, owner and initial_balance

• 1.2. Implement a property with only a getter for the balance ; and a property having both a getter and a setter for the owner.

With your class it must be possible to execute the following scenario:

• 1.2. Implement the __str__() magic in class BankAccount to return a f-string stating the owner and current balance. Loop over all accounts to print them.

• 1.3. Implement these methods :

  • _credit(value) that credits the current account with the value passed in parameter. We will explain the goal of the initial underscore later.
  • transfer_to(recipient, value) that transfers the value passed in parameter to the recipient passed in parameter

• 1.4. Run the following scenario and check that end balances are right:

  • 1.4.1. Alice buys $100 of goods at Walmart
  • 1.4.2. Bob buys $100 of goods at Walmart
  • 1.4.3. Alice makes a donation of $100 to Bob
  • 1.4.4. Bob buys $200 at Walmart

Part 2: The blocked account

Bob is currently overdrawn. To prevent this kind of situation, its customer adviser prefers to convert his account into a blocked account. This way, any purchase would be refused if Bob had not enough money.

• 2.1. Create the InsufficientBalance exception type inheriting from ValueError

• 2.2. Implement a class BlockedBankAccount so that:

  • the BlockedBankAccount inherits from BankAccount. Make sure you do not forget to call parent method with the super() keyword if necessary
  • the transfer_to methods overrides the parent method, with the only difference that it raises InsufficientBalance if the balance is not sufficiently provided to execute the transfer

• 2.3. Replace Bob's account by a blocked account and check that the previous scenario actually raises an exception

• 2.4. Protect the portion of code that looks coherent with try..except in order to catch the exception without interrupting the script

• 2.5. Explain the concept of protected method and the role of the underscore in front of the method name ; and why it is preferable that _credit is protected

Part 3: The account with agios

In real life another kind of account exists: the account whose balance can actually be negative, but it that case the owner must pay agios to his(her) bank.

The proposed rule here is that, when an account is negative after an outgoing money transfer, each day will cost $1 to the owner until the next money credit.

To do so, we need to introduce transaction dates in our simulation.

• 3.1. Implement a class AgiosBankAccount so that:
  • the AgiosBankAccount inherits from BankAccount. Make sure you do not forget to call parent method with the super() keyword if necessary
  • the constructor of this account takes in parameter the account of the bank so that agios can be credited on their account.
  • the transfer_to method overrides the parent method:
    • it takes the transaction_date in parameter, of type datetime
      (Your IDE must warn you about unmatching method signatures: update the other classes to propagate the date to same-name methods)
    • it records the time from which the balance becomes negative. You need an additional attribute for this.
  • the _credit method overrides the method from the parent class, with the only difference that it computes the agios to be payed to the bank and transfer the money to the bank. Round agios to integer values.

• 3.2. Move the code computing the agios in a private method named __check_for_agios, explain the concept of private method and the role of the double underscore
• 3.3. Check your implementation with the previous scenario: After Bob has a negative balance, Alice makes him a transfer 5 days later: make sure that $5 of agios are payed by Bob to his bank.

Part 4: The account package

We have just implemented a very simple tool simulating transactions between bank accounts in Object-Oriented Programming.

In order to use it with a lot of other scenarii and actors, we are going to structure our code within a Python package.

We will organise our accounts with the following terminology:

• bank-internal accounts do not create agios and are not blocked, there are BankAccount and only banks can own such account
• bank-external accounts are for individuals or companies, they can be either blocked or agios accounts.

We would like to be able to import the classes from than manner:

• 4.1. Re-organize your code in order to create this hierarchy of empty .py files first as on the figure.
  Create an empty script scenario1.pyfor the scenario.

• 4.2. Create a logger for each module: agios.py, blocked.py, internal.py. Make sure you log useful debug info in the next questions.

• 4.3. Move the class declaration of AgiosBankAccount in agios.py

• 4.4. Move the class declaration of BlockedBankAccount in blocked.py

• 4.5. Move the class declaration of BankAccount in internal.py

• 4.6. Move the scenario (i.e. the successive instanciation of all accounts of companies and individuals) in scenario1.py

• 4.7. Check each module and add missing relative import statements
  Relative imports start with . or ..

• 4.8. Check each module and add missing absolute import statements such as datetime

Import statements in the scenario must not be relative because scenario1.py will be located outside package account.

• 4.9. Add empty __init__.py files to all directories of the package.

• 4.10. Execute the scenario and check that it leads to the same result as before this refactoring

Part 5: Test your package with pytest and hypothesis

• 5.1. Install pytest with pip
• 5.2. Create independant test files tests/<module>.py for each module of your package
• 5.3. Add an entry in sys.path pointing to the parent folder of your package so that pytest is able to locate and import your account package (*)
• 5.4. using the pytest doc, implement some example-based unit tests for your classes and run the tests with pytest
• 5.5 Using the hypothesis doc, implement some property-based unit tests

(*) Note: This workaround is not ideal since this path is different on each system, and the situation will be fixed once the package will be made installable in Part 6.

Part 6: Automate package building and testing with tox

6.1. Make your package installable

Refer to the doc about package creation and make your package installable. Move your code into a new src/ directory as recommanded by the doc.

Create a dynamic metadata file pyproject.toml and update its metadata (package name, author, dependencies if any).

Delete the sys.path workaround in tests/, since the package is now installable.

Do not run the tests now, since the package is not yet installed in your venv.

6.2. Install, configure and run tox

Refer to the tox basic example.

Create a basic tox.ini so that your package is built and tested against Python 3.12 + 3.11.

Install and run tox in your project. Make sure all tests pass in both environments.

Re-organise your project structure as on the figure.

In Pycharm File > Settings > Project > Project Structure, identify src as a source folder so that the linter can identify your source files.

Part 7: Distribute your package on TestPyPi

• 7.1. Refer to the doc about package creation to create a minimal pyproject.toml
• 7.2. Name your package accounts-<MYNAME> and substitute your name
• 7.3. Install build, wheel and twine
• 7.4. Refer to the doc to build sdist and bdist_wheel distributions
• 7.5. Upload both distributions to TestPyPI using login __token__. As the password, ask for a token or create your own TestPyPI account & new token
• 7.6. Make sure you can install your package from the TestPyPI index via pip:
  pip install accounts-MYNAME --index-url https://test.pypi.org/simple/

Part 8: Implement extra features and distribute a new version on TestPyPI (Optional)

Questions 8.1, 8.2, 8.3 and 8.4 are independant from each other. Choose the most interesting ones and publish a new version of your package with question 8.5.

8.1. Define an Abstract Base Class (ABC) for bank accounts

Since BankAccount is meant to be subclassed, a better design option would be to:

• Transform BankAccount as an ABC in module account.abc
• Implement the ABC with a concrete class InternalBankAccount for the internal use of banks in module account.internal

Refer to the abc documentation to implement and test this new design.

8.2. Write a decorator

In a new module account.monitor, implement a @monitor decorator for the transfer_to methods to print a warning if this user has never transferred an amount of money higher than value before.
- Recall that it is possible to assign an attribute to a function
- Recall that a decorator takes a function in input and returns a function
- Recall that decorators are not bound to class instances

8.3. Override magic methods

Implement the magic method __add__(self, other) so that accounts can be added if they share the same owner, resulting in a new account with the sum of balances.

8.4. Plot the balance history of all accounts

For each account, keep a record of the balance between each transaction and their associated date.

Install matplotlib in your venv and show a single graph at the end of the simulation with a plot for each account owner.

8.5. Publish a new version of your package

If you have implemented the plots, add a new dependency to matplotlib in pyproject.toml

In pyproject.toml set the version to 0.0.2. Drop former distribution versions from dist/ so that you do not upload them again.

Build and publish this new version on TestPyPI.

Exercise 3: Optimization – Bread-First Search in a graph

BFS browses a tree data structure. It starts at the tree root and visits all nodes at the present depth prior to moving on to the nodes at the next depth level.

From Breadth-first_search, Wikipedia.

Black = visited ; Grey = queued to be visited later
BFS is known as parcours en largeur in French

Here is the general algorithm of the BFS in pseudocode:

We have implemented here a naive implementation of the preceeding BFS pseudocode, where a graph is implemented as:

1. : Use a graphical profiler to identify the culprints of this naive code:

• Install snakeviz with pip. Read the documentation.
• Generate a profile for your naive implementation of BFS
• Find the 2 culprints

2.: Implement a new version improved_bfs() that fixes the performance issues that you spotted with the profiler.

3. (Optional): Download Pypy and run the BFS against Pypy instead of CPython. Since Pypy do not support all Python features, you will have to adapt the code with a trial-and-error approach.

Exercise 4: Optimization – Code breaker

MD5 is a message-digest algorithm that is no longer considered safe since 2004 when a research team managed to provoke collisions with MD5.

However it is still widely used in various applications. A common pattern is the use of the MD5 digest of passwords to store them in databases.

In this project, you have extracted MD5 digests from websites accepting only alphabetical (digits excluded) characters of different lengths:

• d0eedb799584d850fdd802fd3c27ae34 (length = 3)
• 9fcce10c03dc2eaada4c361c508c4ebe on a website accepting (length = 4)
• 8b1a9953c4611296a827abf8c47804d7 (length = 5)

Part 1: Use a profiler to identify where time is wasted

Download the script naive_code_breaker.py that breaks MD5 sums.
It reverses the md5 function with hints such as the password characters and size.

This script is highly unoptimized.

• Install snakeviz with pip. Read the documentation.
• Generate a profile for the execution of the script and open it in snakeviz
• Where does you code spends most of the CPU time?

This script is HIGHLY unoptimized and may take long minutes to run with default arguments. Do not waste your time, run it on a lower password length at the beginning:

Part 2: Code optimization

Now try to optimize this naive code. Use the results of the profiler to help.

There are at least 4 points of improvements.

Here are the orders of magnitude of execution duration you must reach:

Do not use multiprocessing here (but we could).

Exercise 5: Async programming

A: Practice coroutines in an empty script

1. Create a coroutine1 sending an event "go 1" after 5 seconds of wait
2. Create a coroutine2 waiting for "go 1" and printing that it received "go 1"
3. Create a coroutine3 sending an event "go 2" after 3 seconds of wait
4. Create a coroutine4 waiting for "go 2" and printing that it received "go 2"
5. In coroutine4, when "go 2" is received, cancel the execution of coroutine2
6. Using time.now() in coroutine4, measure the elapsed waiting time
7. Transfer the wait time into the main coroutine:
   • First, via its returned value
   • Second, via a dedicated future

B: The Chess master

In part B, we will simulate moves of chess during a tournament in which a unique chess master faces many opponents by turns

We will simulate only the timeslots for each move and each player, but we will not simulate the pieces that they play nor their actual outcome.

In a tournament, many games are played in parallel on many boards. The I/O resource that slows down the games here is the time allotted to think-and-play.

Laws of the simulation:

• Master & opponents cannot think-n-play against several players at a time
• Master & opponents cannot think-n-play if the other has not finished his move

B1. Synchronous simulation

Create a synchronous version of the simulator in which only 1 single move can be played at a time in the entire crowd of players. Implement it this way:

• A Chessmaster class with method think_and_play(self, round: "Round", opponent: "Player") that waits 1 sec to simulate the player's move
• A Round class with method play(self, opponent_id: int) that allows the player and then the master to think-and-play.
• A Player class with method think_and_play(self, round: "Round", opponent: "Player") that waits 5 secs to simulate the player's move
• A Simulator class with a single function main() being the entry point

Run a simulation between 3 players against the master, playing 2 rounds each.
Report the overall simulation time (see figure next page).

B2. Asynchronous simulation

In the synchronous version IOs were waited for return every time but they can be optimized: several players can play their move on their board at a time, as long as the laws of the simulator here above are met.

1. Transform relevant functions in async coroutines and await them. Start the event loop with asyncio.run(entry_point). This is still a sequential run.
2. Identify which coroutines will need to be concurrent in the final run.
3. Create tasks for the latter so that their are scheduled by the event loop, and await for their termination. This new step is a full-concurrent run
4. Choose and use several synchronization primives that will enforce the two simulation laws in any case, so that tasks wait the right time to run.

Run the simulation with the same parameters (see figure next page).

Exercise 6: Virus spread simulator

This project consists into simulating the evolution of people contaminated by a disease that spreads according to a simplistic exponential contamination model.

It relies on the R0 variable: it describes the slope of the contamination:

• R0 < 1 = In average, a person contaminates less than 1 person: The number of contamination dicreases
• R0 > 1 = In average, a person contaminates more than 1 person: The number of contaminations increases

The goal is to simulate contamination scenarii and plot them as presented on the next figure, by using a Python iterator.

Description of the model to implement

In our simplistic model:

• The R0 has discrete values for different periods:
  • a value during migitation measures (e.g. lockdowns ; low R0)
  • a value during celebrations (e.g. Christmas ; high R0)
  • a regular value for all other periods
• Time is measured in number of days from day 0, for a certain simulation duration
• Lockdown is triggered automatically when the number of contaminations goes beyond a critical threshold, causing the R0 to decrease.

Bold values are taken as parameters of the simulation.

Part 1: Core of the simulator

1. Implement a class with a constructor with the parameters of the simu, e.g:

1. Store the previous parameters as attributes, and also define the following:

• self.current_date that corresponds to the current date (from 0 to 300)
• self.previous_num_cases that corresponds to the previous number of cases (here, we need a number of 1, corresponding to the very first positive case)

2. Make this class an iterator by implementing:

• __iter__() returns self since the class instance itself is already an iterator

• __next__() must check if we have reached the last day of the simulation:

  • If yes, raise StopIteration
  • If no, update the relevant attributes and return an integer corresponding to the new number of cases = the_regular_r0 * previous_cases_number

3. Create a new class Scenario. Make it an iterable by implementing __iter__() returning an instance of the previous iterator and __len__() returning its length.

4. Create a function plot(iterable) (NOT a class method), that accepts iterable in input and plots it. This function must:

   • Generate all values until the end of the iterator. For this, simply cast the iterator into a list, which which trigger the full generation
   • Store the list of simulated cases into cases_history
   • Import pyplot, call pyplot.plot(xvalues, yvalues) and pyplot.show()
     Refer to the matplotlib documentation if needed

5. Initialize an instance of Scenario and plot it using your plot() function.

Outcome: Now you must see an exponential curve in your plot.

Part 2: Lockdown implementation

1. In __next__(), check if the cases has reached the critical threshold. If yes, substitute the regular R0 by the "lockdown" R0

2. Revert to the regular R0 when number of cases is below the critical level again

Outcome: You should see an exponential curve ending in shaky oscillations around the critical threshold. This is what would happen if the lockdowns were released too fast.

3. Add an attribute counting the number of remaining lockdown days since it has been triggered. Change the behaviour so that lockdown is released only when lockdown_duration days have passed

Outcome: You should see 3 peaks of ascending & descending exponential curves.

Part 3: Simulate a long period and split the plots

Our iterator must generate an illimited number of values.

1. Increase the simulation duration to 950 days.
   It stills works but we want to split the plot in several subplots of 300 days each

2. Overload operator __getitem__(self, item) that is called when you access an iterable slice iterable[start:stop]:

   • That operator must iterate between start and stop indices and return a new list in which elements are generated with function next(iterator).
   • Make sure that you loose no simulated data in case of StopIteration
   • Tip: item is a slice instance that offers item.start and item.stop

3. In plot(), implement a loop that:
   • Defines a plot windows between day start and stop
   • Generates a slice of the iterable
   • Plots the corresponding slice
   • Breaks the loop if no more data is available

Outcome: every time you close the plot window, the plot is refreshed with the next 300 values from the iterator

Part 4: Celebrations (Optional)

In this exercise, celebrations simulate a temporary decrease of the respect of mitigation measures, causing a temporary increase of the R0.

1. In __next__(), simply multiply the current R0 by R0["high"] if this day is present in the list of celebration days.

NB: We use R0 multiplication instead of a substitution of the R0 so that the effect of mitigation measures are still visible but their impact is lowered due to the celebration.

2. In plot(), draw a red vertical bar with pyplot.plot() for each celebration day

Outcome: You must see peaks of cases due to the celebrations. Severity of peaks is highly influenced by the severity of cases at the moment of the celebration.

Part 5: Typing and protocols

We will improve the virus spread simulator with explicit type hints.

1. Start from the solution of the Virus Spread simulator
2. Add type hints to all:
   • Parameters of all functions and methods
     (you may need the Iterator type from typing module for iterators)
   • Class attributes
3. Pay attention to the results of the type checking of your IDE, or install/run mypy to launch type checking.
4. Identify where and why a type error exists.
5. Implement a protocol named Plotable that is implemented by Scenario.
6. Make sure type checking now fully succeeds.

Exercise 7: FastAPI: The TODO app API

You will build a FastAPI application with asynchronous endpoints to manage TODOs. The app will include:

• An in-memory database for TODOs
• Endpoints to list, create, and update TODOs
• A proxy endpoint to fetch a TODO from JSONPlaceholder and enrich it with the user's name from JSONPlaceholder users
• Swagger UI for interactive API documentation

Expected API endpoints of your API

• GET /todos: Returns a list of TODOs
• POST /todos: Creates a new TODO and returns it
• GET /todos/named/{todo_id}: Returns a TODO from JSONPlaceholder with the user's name
• PATCH /todos/{todo_id}: Updates a TODO's completion status and returns the updated TODO

1. Setup and app skeleton

Install fastapi uvicorn httpx into a venv.

Create a main.py and create a FastAPI() app.

Create a first asynchronous endpoint that serves {"message": "Welcome to the FastAPI TODO App!"} for route "/" (root).

Run the app with Uvicorn: uvicorn main:app --reload

Open http://127.0.0.1:8000 in your browser

2. Create the Todo Pydantic model

Create a file named models.py.
Define a Todo class inheriting from pydantic.BaseModel with the following fields:

• userId: integer
• id: integer
• title: string
• completed: boolean

3. Create a in-memory "database"

In main.py, add a list named todos_db to store TODOs. Populate it with two sample TODOs:

4. Implement GET /todos

Add an asynchronous endpoint to list all TODOs. It should:

• Return the todos_db list
• Use the Todo model as the response model @app.get("/todos", response_model=list[Todo])

Test the endpoint at http://127.0.0.1:8000/todos

5. Implement POST /todos

Add an endpoint to create a new TODO. It should:

• Accept a Todo object as input
• Append the new TODO to todos_db: append(todo.dict())
• Return the created TODO

Test the endpoint using Swagger UI at: http://127.0.0.1:8000/docs

6. Proxy the request to JSON placeholder

GET /todos/named/{todo_id}

Add a userName: str field to the model.

Add an endpoint to fetch a TODO from JSONPlaceholder and enrich it with the user's name. It should:

• Accept a todo_id as a path parameter.
• Use httpx.AsyncClient to fetch the TODO from https://jsonplaceholder.typicode.com/todos/{todo_id}.
• Fetch the user from https://jsonplaceholder.typicode.com/users/{userId}.
• Add the user's name to the TODO and return the enriched TODO.

7. Implement PATCH /todos/{todo_id}

Add an endpoint to update a TODO's completion status. It should:

• Accept a todo_id and a completed boolean as input
• Find the TODO in todos_db and update its completed status
• Return the updated TODO
• Raise a 404 error if the TODO is not found