Broker API HubMenu

The finish line, after stage 4 and lesson 31

Capstone: build a read only daily report on the practice API

The project at the end of the path. Build a daily report on the practice API: a program that only reads, paces its requests, handles refusals, tests itself and writes what it found to a file.

Independent and unofficialNo broker account neededReference solution run

Risk warning: CFDs and spread bets are complex instruments and come with a high risk of losing money rapidly due to leverage. You can lose money on these products, and each provider publishes its own loss figure under FCA rules. You should consider whether you understand how they work and whether you can afford to take the high risk of losing your money. Automated and API trading can amplify losses through bugs, connectivity failures and runaway orders.

This project teaches programming. Everything it reads is invented, and nothing it prints is a fact about any market, broker or account, or a suggestion about trading.

The lessons each taught one thing. The capstone asks you to put them together without a script to copy. You are given a brief, a list of addresses and a list of things your program must do. How you arrange the code is up to you.

The program is a daily report: the kind of small tool that runs each morning, reads an account and writes down what it found. It talks to this site's practice API, so it needs no account and no key, every value it reads is invented, and it cannot place a trade. It only ever reads.

It produces two files: a short report, and a list of closed trades with a running total. Write your own before you open the reference solution. That solution is one way to meet the brief, not the answer. If yours meets every criterion below, it is finished.

The practice API addresses you are given

Practice data, not any broker's API: fixed files of invented values on this site. Each opens in a browser tab. Open them and read them before you write anything.

The closed trades arrive in pages. Start at the first page and follow the next field of each page until it is null. Every address on the practice API, and what it cannot do

The brief

Write one Python program that does these things, in this order.

  1. Read your settings from the environment in one place: the address of the practice API and the pause between requests, each with a default. The program needs no secret, so it reads none.
  2. Send every request through one place in your code. That place leaves the pause before every request but the first, logs what was asked and what came back, refuses a reply that does not say it is practice data, and handles a refusal: wait and try again for a rate limit or a server fault, three attempts at most, and stop at once for anything else.
  3. Fetch the account summary, the open positions and the working orders.
  4. Collect every page of closed trades by following each page's pointer to the next, and put the trades in closing order, oldest first.
  5. Work out the running profit of each position from its levels, the exposure on each instrument with the gross and net totals, the profit or loss of the closed trades for each instrument, the highest running total and the worst drop from a high.
  6. For every instrument that has a position or a working order, fetch its price history and put the latest close beside its five day average.
  7. Write the report to a text file and the closed trades to a CSV file with a running total, and print the report.
  8. Write tests with unittest for your own functions, on replies you have saved or made up, and run them before the first request.

Acceptance criteria

Your program is finished when every line here is true of it. Check each one yourself. Where you can, check it with an assert.

  1. Every request is a GET to the practice API. The program places, changes and cancels nothing, and it reads no key, password or token.
  2. A pause is left before every request but the first, and the report says how many requests were made.
  3. A reply that carries an error is never used as data. A rate limit is retried after the wait the reply asks for, three attempts at most. An expired session is not retried. A test proves each of these.
  4. Every page of trades is collected and the loop stops by itself. The number collected equals the number the API says there are and no trade appears twice. The program refuses to carry on otherwise.
  5. No money value is held as a float. The running profit you work out equals the figure in the account summary, and the last running total in the CSV file equals the total in the report.
  6. Each average is made only from closes up to the latest day, and the report says in words that it is arithmetic and not a signal.
  7. The tests run before the first request and none of them uses the network. If a test fails, the program stops without making a request.
  8. The program says what it did as it goes, with logging. If it cannot finish, it logs why, writes no report and hands the terminal a number other than 0.
  9. Running the program twice gives the same two files. It replaces them and never appends.

Lessons this project draws on

If a part of the brief stops you, the lesson that taught it is the place to go back to.

A reference solution

Open this after you have written your own. It is one way to meet the brief. The code is the file that was run, and the output beneath it is what that run printed.

The reference solution
# Capstone, reference solution: a read-only daily report built on the practice API.
# It paces its calls, handles refusals, tests itself before it touches the network and writes two files.
# One way to meet the brief, not the only way.
# The practice API is a set of fixed files of invented values. It is not any broker's API.
# It needs no account and no key, and this program only ever reads.
import csv
import json
import logging
import os
import sys
import time
import unittest
import urllib.error
import urllib.request
from decimal import Decimal
from unittest import mock

PENNY = Decimal("0.01")
AVERAGE_DAYS = 5
MAX_ATTEMPTS = 3
MAX_PAGES = 20
REPORT_FILE = "daily-report.txt"
TRADES_FILE = "daily-trades.csv"

log = logging.getLogger("daily-report")
log.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
log.addHandler(handler)


class ReportStopped(Exception):
    """The report cannot be finished. The message says why."""


# ---------------------------------------------------------------- settings
def load_settings(environment):
    """Everything that may differ from one machine to the next, read in one place. No secret is needed."""
    text = environment.get("PRACTICE_PAUSE_SECONDS", "1")
    try:
        pause = float(text)
    except ValueError:
        raise ReportStopped("PRACTICE_PAUSE_SECONDS must be a number, not " + repr(text)) from None
    if pause < 0:
        raise ReportStopped("PRACTICE_PAUSE_SECONDS cannot be less than 0")
    return {"base": environment.get("PRACTICE_API", "https://openbrokerapi.com/mock/v1").rstrip("/"), "pause": pause}


# ---------------------------------------------------------------- one place where requests are made
def make_fetch(base):
    def fetch(path):
        """One GET. Returns the status and the body, whether the server was pleased or not."""
        request = urllib.request.Request(base + path, headers={"Accept": "application/json", "User-Agent": "capstone-daily-report"})
        try:
            with urllib.request.urlopen(request, timeout=10) as reply:
                status, body = reply.status, json.load(reply, parse_float=Decimal)
        except urllib.error.HTTPError as refusal:
            try:
                return refusal.code, json.load(refusal)
            except ValueError:      # the refusal came with a body that is not JSON
                return refusal.code, {"error": {"status": refusal.code, "code": "http_" + str(refusal.code)}}
        if "error" in body:         # the practice API always sends 200 and puts the real status in the body
            status = body["error"]["status"]
        return status, body
    return fetch


def decide(status, body):
    """What to do about one reply: an action and, for a retry, how many seconds to wait first."""
    if status < 400:
        return "use it", 0
    if status == 429:
        return "wait and retry", body["error"].get("retryAfterSeconds", 5)
    if status in (500, 502, 503, 504):
        return "wait and retry", 5
    if status == 401:
        return "sign in again", 0
    return "stop and fix the request", 0


class Api:
    """Every request goes through get, so pacing, logging and error handling happen in one place."""

    def __init__(self, fetch, sleep, pause):
        self.fetch, self.sleep, self.pause = fetch, sleep, pause
        self.requests = 0

    def get(self, path):
        for attempt in range(1, MAX_ATTEMPTS + 1):
            if self.requests:
                self.sleep(self.pause)          # a gap before every request but the first
            self.requests += 1
            status, body = self.fetch(path)
            action, seconds = decide(status, body)
            if action == "use it":
                if body.get("practice") is not True:
                    raise ReportStopped(path + " did not say it was practice data")
                log.info("GET %s status %s", path, status)
                return body
            log.warning("GET %s status %s %s: %s", path, status, body["error"]["code"], action)
            if action != "wait and retry":
                raise ReportStopped(path + " was refused with status " + str(status) + ": " + action)
            if attempt < MAX_ATTEMPTS:
                self.sleep(seconds)
        raise ReportStopped(path + " was still refused after " + str(MAX_ATTEMPTS) + " attempts")


def collect_trades(api):
    """Follow each page's pointer to the next, and prove that nothing is missing or repeated."""
    trades, visited, path = [], [], "/trades/page-1.json"
    while path is not None:
        if path in visited or len(visited) == MAX_PAGES:
            raise ReportStopped("the trade history does not end: stopped at " + path)
        answer = api.get(path)
        visited.append(path)
        trades.extend(answer["trades"])
        path = answer["paging"]["next"]
    ids = [trade["tradeId"] for trade in trades]
    if len(set(ids)) != len(ids):
        raise ReportStopped("the same trade arrived twice")
    if len(trades) != answer["paging"]["totalRows"]:
        raise ReportStopped("expected " + str(answer["paging"]["totalRows"]) + " trades, collected " + str(len(trades)))
    return sorted(trades, key=lambda trade: trade["closedAt"])      # oldest first


# ---------------------------------------------------------------- the sums
def money(value):
    return Decimal(value).quantize(PENNY)


def exposure(position):
    """Size times the price the position would close at. BUY counts up, SELL counts down."""
    if position["direction"] == "BUY":
        return position["size"] * position["bid"]
    return -(position["size"] * position["offer"])


def running_profit(position):
    if position["direction"] == "BUY":
        return (position["bid"] - position["openLevel"]) * position["size"]
    return (position["openLevel"] - position["offer"]) * position["size"]


def totals_by(rows, key, value):
    """Add up value(row) for each different row[key]."""
    totals = {}
    for row in rows:
        totals[row[key]] = totals.get(row[key], Decimal("0")) + value(row)
    return totals


def worst_drop(profits):
    """The highest running total, and the largest fall from a high to a later low."""
    running = peak = drop = Decimal("0")
    for profit in profits:
        running += profit
        peak = max(peak, running)
        drop = max(drop, peak - running)
    return peak, drop


def moving_average(closes, days):
    """The average of the last `days` closes, or None if there are not enough."""
    if len(closes) < days:
        return None
    return sum(closes[-days:]) / days


# ---------------------------------------------------------------- tests, on made up replies, with no network
def reply(extra):
    return 200, {"practice": True, **extra}


REFUSED_429 = (200, {"practice": True, "error": {"status": 429, "code": "rate_limited", "retryAfterSeconds": 30}})
REFUSED_401 = (200, {"practice": True, "error": {"status": 401, "code": "session_expired"}})


def scripted(answers):
    """A stand in for fetch that hands back prepared answers in order and remembers what was asked."""
    asked = []

    def fetch(path):
        asked.append(path)
        status, body = answers[len(asked) - 1]
        return (body["error"]["status"] if "error" in body else status), body
    return fetch, asked


class ReportTests(unittest.TestCase):
    def setUp(self):
        blocker = mock.patch("urllib.request.urlopen", side_effect=AssertionError("a test tried to use the network"))
        blocker.start()
        self.addCleanup(blocker.stop)
        quiet = mock.patch.object(log, "disabled", True)     # the tests cause warnings on purpose
        quiet.start()
        self.addCleanup(quiet.stop)
        self.waits = []

    def api(self, answers):
        fetch, asked = scripted(answers)
        return Api(fetch, self.waits.append, 1), asked

    def test_exposure_counts_a_sell_as_negative(self):
        self.assertEqual(exposure({"direction": "BUY", "size": 2, "bid": Decimal("50.5"), "offer": Decimal("51")}), Decimal("101.0"))
        self.assertEqual(exposure({"direction": "SELL", "size": 2, "bid": Decimal("50.5"), "offer": Decimal("51")}), Decimal("-102"))

    def test_moving_average_needs_enough_closes(self):
        self.assertEqual(moving_average([Decimal(1), Decimal(2), Decimal(6)], 2), Decimal(4))
        self.assertIsNone(moving_average([Decimal(1)], 2))

    def test_worst_drop_is_measured_from_the_high_before_it(self):
        self.assertEqual(worst_drop([Decimal(10), Decimal(-4), Decimal(-3), Decimal(20), Decimal(-1)]), (Decimal(23), Decimal(7)))

    def test_a_rate_limit_is_waited_out_and_retried(self):
        api, asked = self.api([REFUSED_429, REFUSED_429, reply({"account": {}})])
        self.assertIn("account", api.get("/account.json"))
        self.assertEqual(len(asked), 3)
        self.assertEqual(self.waits, [30, 1, 30, 1])     # the wait the reply asked for, then the usual gap

    def test_a_rate_limit_that_never_lifts_is_given_up_on(self):
        api, asked = self.api([REFUSED_429] * 5)
        with self.assertRaises(ReportStopped):
            api.get("/account.json")
        self.assertEqual(len(asked), MAX_ATTEMPTS)

    def test_an_expired_session_is_not_retried(self):
        api, asked = self.api([REFUSED_401, reply({})])
        with self.assertRaises(ReportStopped):
            api.get("/account.json")
        self.assertEqual(len(asked), 1)

    def test_a_reply_that_is_not_practice_data_is_refused(self):
        api, asked = self.api([(200, {"account": {}})])
        with self.assertRaises(ReportStopped):
            api.get("/account.json")

    def test_pages_are_followed_and_put_oldest_first(self):
        page = lambda ids, following: reply({"trades": [{"tradeId": i, "closedAt": i} for i in ids], "paging": {"next": following, "totalRows": 3}})
        api, asked = self.api([page(["c", "b"], "/trades/page-2.json"), page(["a"], None)])
        self.assertEqual([trade["tradeId"] for trade in collect_trades(api)], ["a", "b", "c"])
        self.assertEqual(asked, ["/trades/page-1.json", "/trades/page-2.json"])

    def test_a_short_history_is_refused(self):
        api, asked = self.api([reply({"trades": [{"tradeId": "a", "closedAt": "a"}], "paging": {"next": None, "totalRows": 2}})])
        with self.assertRaises(ReportStopped):
            collect_trades(api)


def run_tests():
    suite = unittest.defaultTestLoader.loadTestsFromTestCase(ReportTests)
    result = unittest.TestResult()
    suite.run(result)
    for test, why in result.failures + result.errors:
        log.error("Test failed: %s\n%s", test.id().split(".")[-1], why)
    return result.testsRun, len(result.failures) + len(result.errors)


# ---------------------------------------------------------------- the report
def build_report(api):
    account = api.get("/account.json")
    positions = api.get("/positions.json")["positions"]
    orders = api.get("/orders.json")["orders"]
    trades = collect_trades(api)

    profit = sum(running_profit(position) for position in positions)
    if profit != account["account"]["profitLoss"]:
        raise ReportStopped("the positions and the account summary disagree about the running profit")
    by_instrument = totals_by(positions, "instrumentId", exposure)
    closed = totals_by(trades, "instrumentId", lambda trade: trade["profitLoss"])
    winners = len([trade for trade in trades if trade["profitLoss"] > 0])
    losers = len([trade for trade in trades if trade["profitLoss"] < 0])
    peak, drop = worst_drop([trade["profitLoss"] for trade in trades])

    summary = account["account"]
    lines = [
        "Daily report for " + summary["name"] + ", as of " + account["asOf"],
        "Practice data: every value is invented. Read only: nothing was placed, changed or cancelled.",
        "",
        "Account",
        "  Balance " + str(money(summary["balance"])) + ", equity " + str(money(summary["equity"])) + ", available " + str(money(summary["available"])) + " " + summary["currency"],
        "  Running profit or loss " + str(money(profit)) + ", which agrees with the open positions",
        "",
        "Exposure by instrument, from " + str(len(positions)) + " open positions",
    ]
    for name in sorted(by_instrument):
        lines.append("  " + name.ljust(7) + str(money(by_instrument[name])).rjust(11))
    lines.append("  Gross " + str(money(sum(abs(value) for value in by_instrument.values()))) + ", net " + str(money(sum(by_instrument.values()))))
    lines += ["", "Working orders: " + str(len(orders))]
    for order in orders:
        lines.append("  " + order["orderId"] + " " + order["direction"] + " " + str(order["size"]) + " " + order["instrumentId"] + " " + order["type"] + " at " + str(order["level"]))
    lines += ["", "Closed trades: " + str(len(trades)) + ", " + str(winners) + " winners and " + str(losers) + " losers, from " + trades[0]["closedAt"][:10] + " to " + trades[-1]["closedAt"][:10]]
    for name in sorted(closed):
        lines.append("  " + name.ljust(7) + str(money(closed[name])).rjust(11))
    lines.append("  Total " + str(money(sum(closed.values()))))
    lines.append("  Highest running total " + str(money(peak)) + ", worst drop from a high " + str(money(drop)))

    lines += ["", "Latest close beside its " + str(AVERAGE_DAYS) + " day average. Arithmetic only, not a signal."]
    for name in sorted(set(by_instrument) | {order["instrumentId"] for order in orders}):
        history = api.get("/history/" + name + ".json")
        closes = [candle["close"] for candle in history["candles"]]
        lines.append("  " + name.ljust(7) + str(money(closes[-1])).rjust(11) + str(money(moving_average(closes, AVERAGE_DAYS))).rjust(11))
    lines += ["", "Requests made: " + str(api.requests) + ", every one a GET"]
    return lines, trades


def write_files(lines, trades):
    with open(REPORT_FILE, "w", encoding="utf-8", newline="\n") as f:        # "w" replaces the file, it never appends
        f.write("\n".join(lines) + "\n")
    running = Decimal("0")
    with open(TRADES_FILE, "w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f, lineterminator="\n")
        writer.writerow(["tradeId", "closedAt", "instrumentId", "direction", "size", "profitLoss", "runningTotal"])
        for trade in trades:
            running += trade["profitLoss"]
            writer.writerow([trade["tradeId"], trade["closedAt"], trade["instrumentId"], trade["direction"], trade["size"], money(trade["profitLoss"]), money(running)])


def main():
    run, failed = run_tests()
    log.info("Tests: %d run, %d failed, none of them used the network", run, failed)
    if failed:
        log.error("Stopped before making any request, because a test failed")
        return 1
    try:
        settings = load_settings(os.environ)
        log.info("Pause between requests: %s seconds", settings["pause"])
        api = Api(make_fetch(settings["base"]), time.sleep, settings["pause"])
        lines, trades = build_report(api)
        write_files(lines, trades)
    except ReportStopped as problem:
        log.error("No report written: %s", problem)
        return 1
    except (urllib.error.URLError, TimeoutError) as problem:
        log.error("No report written: could not reach the practice API: %s", problem)
        return 1
    log.info("Wrote %s and %s", REPORT_FILE, TRADES_FILE)
    print("\n".join(lines))
    return 0


if __name__ == "__main__":
    sys.exit(main())
What it prints
INFO Tests: 9 run, 0 failed, none of them used the network
INFO Pause between requests: 1.0 seconds
INFO GET /account.json status 200
INFO GET /positions.json status 200
INFO GET /orders.json status 200
INFO GET /trades/page-1.json status 200
INFO GET /trades/page-2.json status 200
INFO GET /trades/page-3.json status 200
INFO GET /history/HBR100.json status 200
INFO GET /history/LNTN.json status 200
INFO GET /history/TIDE.json status 200
INFO Wrote daily-report.txt and daily-trades.csv
Daily report for Practice account, as of 2026-01-05T09:30:00Z
Practice data: every value is invented. Read only: nothing was placed, changed or cancelled.

Account
  Balance 10000.00, equity 10067.50, available 8794.00 GBP
  Running profit or loss 67.50, which agrees with the open positions

Exposure by instrument, from 2 open positions
  HBR100    10114.00
  LNTN     -12416.50
  Gross 22530.50, net -2302.50

Working orders: 2
  O-2001 BUY 3 TIDE LIMIT at 61.5
  O-2002 SELL 1 HBR100 STOP at 4990

Closed trades: 12, 8 winners and 4 losers, from 2025-12-19 to 2026-01-01
  HBR100       20.00
  LNTN         16.00
  TIDE          6.40
  Total 42.40
  Highest running total 44.20, worst drop from a high 42.30

Latest close beside its 5 day average. Arithmetic only, not a signal.
  HBR100     5057.50    5049.90
  LNTN        248.20     248.48
  TIDE         62.60      62.30

Requests made: 9, every one a GET

Run twice. Run with Python 3.12 on Windows against a copy of the practice API files served over HTTP on the same machine, with PRACTICE_API set to the address of the copy. They are the files this site serves. No account and no key. Both runs printed the same lines and wrote the same two files.

The first lines of the CSV file it writes
tradeId,closedAt,instrumentId,direction,size,profitLoss,runningTotal
T-3001,2025-12-19T15:40:00Z,HBR100,BUY,1,16.00,16.00
T-3002,2025-12-19T16:05:00Z,TIDE,BUY,5,1.00,17.00
T-3003,2025-12-22T11:30:00Z,LNTN,SELL,10,14.00,31.00

The file holds a header and 12 lines, one for each closed trade, oldest first.

Your figures should match the report exactly, because the practice API never changes and money held as a Decimal has one right answer. Your layout, your function names and your log lines need not match.

What finishing this means

Finishing this does not make a program fit to trade with. It has only ever read fixed files of invented values, with nothing to sign in to. It shows that you can ask an API for data with care, keep money exact, handle a refusal, prove that your own functions do what you think and leave a record behind. A real API differs in its addresses, its limits and its errors, and it changes while you watch. Any code that talks to a broker belongs on a demo account first.

Your progress

Back to the path

  1. A command line tool
  2. The capstone projectYou are here

Your marks are kept in this browser only, in its local storage. They are not sent anywhere, there is no account, and they will not follow you to another browser or device. To remove them, use Clear my progress on the learning track page, or clear this site's data in your browser's settings. Marking needs JavaScript. With scripts off the track reads the same, without the ticks.