#!/usr/bin/python3
"""Get price candles from cTrader into PostgreSQL."""

import argparse
import base64
import copy
import csv
import fcntl
import getpass
import itertools
import json
import lzma
import math
import os
import pwd
import shlex
import shutil
import subprocess
import struct
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from contextlib import contextmanager
from datetime import UTC, datetime, timedelta
from decimal import Decimal

import psycopg
from psycopg.conninfo import conninfo_to_dict, make_conninfo
from websockets.exceptions import WebSocketException
from websockets.sync.client import connect

SELF = os.path.basename(sys.argv[0])
TTY = sys.stderr.isatty()

DEFAULT_CONFIG = "/etc/ohlcdb/config.yaml"
DEFAULT_TOKEN_FILE = "/var/lib/ohlcdb/tokens.json"
# Where the package puts the config template and the SQL files; a run from
# a checkout finds them beside the script instead.
SHARE = "/usr/share/ohlcdb"
DEFAULT_LOG = "/var/log/ohlcdb/ohlcdb.log"
DEFAULT_EVERY = 15

# Marks the collector's line among the others in its user's crontab. crond
# hands the whole line to sh, which reads the mark as a comment.
CRON_TAG = "# ohlcdb"

# Each environment is served by its own host, and an account is reachable
# only on the one of its environment.
HOSTS = {"live": "live.ctraderapi.com", "demo": "demo.ctraderapi.com"}

# JSON is served only on this port; 5035 speaks protobuf.
PORT = 5036
TIMEOUT = 30

# How far a request may reach is not documented per period, and a request
# that exceeds it is not refused: the server drops the oldest candles without
# saying so. A day's worth of M1 per request, scaled with the period and
# capped at 300 days, stays inside every limit reported.
CANDLES_PER_REQUEST = 1440
MAX_SPAN = timedelta(days=300)

# Historical data requests are limited to 5 per second per connection.
HISTORY_INTERVAL = 0.25

# Refresh a week before expiry, so that a few days of failed runs do not
# leave the collector holding a dead token.
REFRESH_MARGIN = 7 * 86400

ERROR_RES = 50
APPLICATION_AUTH_REQ = 2100
ACCOUNT_AUTH_REQ = 2102
SYMBOLS_LIST_REQ = 2114
SYMBOL_BY_ID_REQ = 2116
GET_TRENDBARS_REQ = 2137
OA_ERROR_RES = 2142
ACCOUNTS_TOKEN_INVALIDATED_EVENT = 2147
CLIENT_DISCONNECT_EVENT = 2148
GET_TICKDATA_REQ = 2145
GET_ACCOUNTS_REQ = 2149
REFRESH_TOKEN_REQ = 2173

# The two quote streams. A candle is built on the priced side, bid for the
# charts, but its volume counts the updates of both.
SIDES = {"bid": 1, "ask": 2}

# A reply holds at most this many ticks, the newest ones of the window, so a
# window that does not fit comes back short of its older part. Windows that
# come back capped are halved, and quiet ones are doubled, up to the week
# the server allows per request.
TICKS_PER_REPLY = 10_000
FIRST_TICK_WINDOW = timedelta(hours=1)
MAX_TICK_WINDOW = timedelta(days=7)
MIN_TICK_WINDOW = timedelta(seconds=1)

# Requests one run spends on ticks. Catching up on a year of them takes
# thousands, so a run does its share and leaves the rest to the next: the
# candles stay minutes behind the market while the ticks fill in.
TICK_BUDGET = 300

# The candle lengths built from the minutes where the source gives none.
# Dukascopy publish M1 and nothing else, and the levels read H1 and D1, so
# without these a feed of theirs has twenty years of candles and not one
# session. What is built is not collected: it can be dropped and made
# again, since the minutes it comes from are the only thing stored that
# nothing else can reproduce.
ROLLUP = ["M5", "M15", "M30", "H1", "H4", "D1", "W1", "MN1"]

# Who we say we are when asking Dukascopy for anything, and not a detail:
# both their API and their archive turn away a client that arrives as
# Python-urllib, the API with a refusal that lasts a minute and a half and
# the archive with one it does not lift at all. Any name of our own is
# enough - theirs is a filter against nameless traffic, not against us -
# and with one the archive answers in seconds where it used to take a
# minute, which is most of what looked like throttling.
DUKAS_AGENT = "ohlcdb (+https://git.snix.me/snix/ohlcdb)"

# Dukascopy serve the same history twice. Their charts read it from an API
# that answers a day of candles as JSON in a fifth of a second, eighty days
# running without a pause and without complaint; their archive of files
# answers the same day in seconds to tens of seconds, and grows slower the
# longer it is asked. So the API is asked first and the files are the
# reserve, for the day the API changes or goes: it is the one their web
# widgets call, published for nobody, and nothing says it will stay as it
# is.
JETTA_CANDLES = "https://jetta.dukascopy.com/v1/candles/minute/{code}/BID/{year}/{month}/{day}"
# Everything they publish, with the code to ask for it by and the scale its
# prices are quoted at. Their code is the name with a hyphen where the
# slash goes - XAU-USD for XAUUSD - which is not worth guessing when they
# will say it.
JETTA_INSTRUMENTS = "https://jetta.dukascopy.com/v1/instruments"
JETTA_INSTRUMENT = "https://jetta.dukascopy.com/v1/instruments/{code}"
JETTA_TIMEOUT = 30
JETTA_RETRIES = 3
# A server having a moment is worth another go at once; a refusal is worth
# one only after it has passed, and measured, theirs lasts a minute and a
# half.
JETTA_PAUSE = 2.0
JETTA_COOLDOWN = 60
# Days running it will ask the API for after it has stopped answering
# before leaving it alone for the rest of the run. Without this, an API
# that has gone costs every single day three tries and their pauses on top
# of the file that has to be fetched anyway.
JETTA_GIVE_UP = 5
# A day the market was shut comes back as an answer with no minutes in it,
# not as an error, so a weekend costs one quick request and never reaches
# the files. Today and any day after it are refused instead, which is why a
# run stops at yesterday.

# The archive: one file per instrument per day, a thousand four hundred and
# forty minutes of it, compressed with plain LZMA. The month in the path
# counts from zero, as in the Java that writes them: August is 07.
DUKAS_URL = "https://datafeed.dukascopy.com/datafeed/{symbol}/{year:04d}/{month:02d}/{day:02d}/BID_candles_min_1.bi5"
# A record is a whole minute: seconds from the day's start, then open,
# close, low, high as whole points, then what was traded.
DUKAS_RECORD = ">5if"
DUKAS_MINUTES = 1440
# The server is not asked who we are, and answers 429 when asked too often.
# How often that is, it does not say, so the pace is theirs to set: a run
# starts here, slows by half as much again at every refusal, and never goes
# below the floor. A refusal also costs a wait, since their limiter stays
# cross for a while - seconds of backing off are not enough, as three
# retries and ten seconds proved.
DUKAS_INTERVAL = 2.0
DUKAS_SLOWEST = 15.0
DUKAS_COOLDOWN = 60
DUKAS_LONGEST_WAIT = 300
DUKAS_RETRIES = 4
# Their free history answers a client that asks too often with either of
# these; the rest of the codes here are a server having a moment, and are
# worth another go all the same.
DUKAS_REFUSALS = (429, 503)
DUKAS_AGAIN = DUKAS_REFUSALS + (500, 502, 504)
# Their server answers a client it is throttling slowly rather than turning
# it away: replies of ten to twenty seconds are normal, and their own
# gateway gives up around a minute. A short timeout turns a slow answer
# into a failure and slows the run for no reason, so this is generous.
DUKAS_TIMEOUT = 75
# Days a run takes when it is not told otherwise. Through the API that is
# three minutes, and a full twenty years - eight and a half thousand days -
# is under half an hour in one go.
DUKAS_BUDGET = 1000
# Where their history comes from, as the database names a feed: the prices
# are real market ones, and SWFX is the marketplace whose trades they are -
# which is worth remembering every time their volume is read.
DUKAS_BROKER = "Dukascopy"
DUKAS_ENVIRONMENT = "live"
DUKAS_ACCOUNT = "SWFX"
# A marketplace, so its volume is an amount that changed hands there. A
# cTrader account is a broker's own book, where the volume of a candle is a
# count of quote updates and nothing was traded at all.
DUKAS_MARKET = "spot"
CTRADER_MARKET = "cfd"
# What a feed's market may be, as the CHECK in schema.sql spells it: the
# only one that is asked for is Databento's, since what they sell depends on
# the dataset, while the other two sources are one kind of market each.
MARKETS = ("futures", "spot", "cfd")
# Who hands the prices over. For these two it is the same name as the
# broker, since the venue is also the shop; it is not always so.
CTRADER_SOURCE = "cTrader"

# Databento sell what the exchanges publish, and are paid by the byte. The
# cost of a request can be asked for free before making it, which is the
# only thing that makes an unattended collector acceptable here: a wrong
# date is a bill rather than a wasted afternoon.
DBENTO_HOST = "https://hist.databento.com/v0"
DBENTO_TIMEOUT = 300
DBENTO_RETRIES = 3
DBENTO_PAUSE = 5.0
# Their prices are integers scaled by a thousand million, and their times
# are nanoseconds: the same in every dataset and every schema, so it is a
# constant here rather than a setting.
DBENTO_SCALE = Decimal(10) ** -9
DBENTO_NANOS = 1_000_000_000
# A month at a time. The price is the same however it is cut - their meter
# runs on the data, not on the number of requests - and a month is thirty
# thousand candles, which fits in memory and shows up in the log as it goes.
DBENTO_SPEND = 1.00
DBENTO_SOURCE = "Databento"
# Where the decimal places land when there are no prices to read them off.
DBENTO_DIGITS = 2
# The month each futures code stands for, written the same way by every
# exchange. The letters skip I, so that it cannot be read as a one.
CONTRACT_MONTHS = {"F": 1, "G": 2, "H": 3, "J": 4, "K": 5, "M": 6,
                   "N": 7, "Q": 8, "U": 9, "V": 10, "X": 11, "Z": 12}
# How a continuous series is written: the roll rule and how far down the
# ladder. GC.v.0 is the front by volume, .n.1 the second by open interest.
CONTINUOUS_RULES = ("v", "n", "c")

# Each period's ProtoOATrendbarPeriod value and its length in minutes. The
# server writes enums as their numbers, not their names, and is only known to
# read them the same way. A month counts as 30 days, which only sizes the
# requests.
PERIODS = {
    "M1": (1, 1), "M2": (2, 2), "M3": (3, 3), "M4": (4, 4), "M5": (5, 5),
    "M10": (6, 10), "M15": (7, 15), "M30": (8, 30),
    "H1": (9, 60), "H4": (10, 240), "H12": (11, 720),
    "D1": (12, 1440), "W1": (13, 10080), "MN1": (14, 43200),
}
SCOPES = {0: "view", 1: "trade"}

# Every run downloads its first candle again, and --since may cover a range
# already stored: the WHERE leaves identical rows alone instead of rewriting
# them into dead tuples.
UPSERT = """
    INSERT INTO bars (pair_id, period, ts, open, high, low, close, volume)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
    ON CONFLICT (pair_id, period, ts) DO UPDATE
    SET open = excluded.open, high = excluded.high, low = excluded.low,
        close = excluded.close, volume = excluded.volume
    WHERE (bars.open, bars.high, bars.low, bars.close, bars.volume)
        IS DISTINCT FROM (excluded.open, excluded.high, excluded.low,
                          excluded.close, excluded.volume)
"""

# A tick never changes, and the key holds everything a tick is, so a window
# fetched again simply adds nothing.
TICK_INSERT = """
    INSERT INTO ticks (pair_id, side, ts, price) VALUES (%s, %s, %s, %s)
    ON CONFLICT DO NOTHING
"""


class Fatal(Exception):
    pass


class NoTokens(Fatal):
    """There is no pair to work with: none stored yet, or the stored one was
    turned down. Nothing but a new pair from the Playground gets past this,
    which only a terminal can be asked for."""


class ApiError(Fatal):
    def __init__(self, payload):
        self.code = payload.get("errorCode", "")
        self.retry_after = int(payload.get("retryAfter", 1))
        super().__init__(f"cTrader: {self.code}: {payload.get('description', '')}")


class Client:
    def __init__(self, ws):
        self.ws = ws
        self.ids = itertools.count(1)

    def request(self, payload_type, payload, attempts=3):
        while True:
            try:
                return self._exchange(payload_type, payload)
            except ApiError as e:
                attempts -= 1
                if e.code != "BLOCKED_PAYLOAD_TYPE" or attempts == 0:
                    raise
                time.sleep(e.retry_after)

    def _exchange(self, payload_type, payload):
        msg_id = str(next(self.ids))
        self.ws.send(json.dumps({
            "clientMsgId": msg_id,
            "payloadType": payload_type,
            "payload": payload,
        }))
        while True:
            msg = json.loads(self.ws.recv(timeout=TIMEOUT))
            kind = msg.get("payloadType")
            body = msg.get("payload", {})
            if kind in (ACCOUNTS_TOKEN_INVALIDATED_EVENT, CLIENT_DISCONNECT_EVENT):
                raise Fatal(f"cTrader closed the session: {body.get('reason', kind)}")
            # Requests go one at a time, so an error without an id is ours too.
            if kind in (ERROR_RES, OA_ERROR_RES) and msg.get("clientMsgId", msg_id) == msg_id:
                raise ApiError(body)
            # Heartbeats and events arrive between the replies; ours is the
            # one carrying our id back.
            if msg.get("clientMsgId") != msg_id:
                continue
            # Every request is answered by the payload type that follows it.
            if kind != payload_type + 1:
                raise Fatal(f"cTrader: expected payload type {payload_type + 1}, got {kind}")
            return body


def log(msg):
    # On a terminal, first wipe the progress line that may be standing there.
    # Anywhere else the line most likely ends up in the log file, which has
    # no other record of when it was written.
    prefix = "\r\x1b[K" if TTY else datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ ")
    print(prefix + msg, file=sys.stderr, flush=True)


def die(msg):
    log(f"ERROR: {msg}")
    sys.exit(1)


def progress(msg):
    print(f"\r{msg}\x1b[K", end="", file=sys.stderr, flush=True)


def duration(seconds):
    m, s = divmod(int(seconds), 60)
    h, m = divmod(m, 60)
    return f"{h}h{m:02d}m" if h else f"{m}m{s:02d}s" if m else f"{s}s"


def value(cfg, path, default=None):
    """The setting at a dotted path, or default where it is missing or null."""
    node = cfg
    for key in path.split("."):
        if not isinstance(node, dict) or node.get(key) is None:
            return default
        node = node[key]
    return node


def put(cfg, path, v):
    *parents, last = path.split(".")
    for key in parents:
        if not isinstance(cfg.get(key), dict):
            cfg[key] = {}
        cfg = cfg[key]
    cfg[last] = v


def need(cfg, path):
    v = value(cfg, path)
    if v in (None, "", 0, []):
        raise Fatal(f"config: {path} is not set; {SELF} config sets it")
    return v


def listed(cfg, path, default):
    v = value(cfg, path, default)
    return [v] if isinstance(v, str) else list(v)


def periods(cfg):
    chosen = listed(cfg, "ctrader.periods", ["M1"])
    unknown = [p for p in chosen if p not in PERIODS]
    if unknown or not chosen:
        raise Fatal(f"config: periods: {' '.join(unknown) or 'none given'}; "
                    f"choose among {' '.join(PERIODS)}")
    return chosen


def rollup(cfg):
    """The candle lengths to build from the minutes for a pair whose source
    does not publish them. M1 is not among them: it is what they are made
    of."""
    chosen = listed(cfg, "rollup", ROLLUP)
    buildable = [p for p in PERIODS if p != "M1"]
    unknown = [p for p in chosen if p not in buildable]
    if unknown:
        raise Fatal(f"config: rollup: {' '.join(unknown)}; choose among {' '.join(buildable)}")
    return chosen


def start_date(cfg, path="ctrader.start"):
    """Where a pair and period with nothing stored begins, if anywhere."""
    v = value(cfg, path)
    if v in (None, ""):
        return None
    # Written by hand without quotes, a date may come back with a time.
    try:
        return datetime.strptime(str(v)[:10], "%Y-%m-%d").replace(tzinfo=UTC)
    except ValueError:
        raise Fatal(f"config: {path} is {v!r}, not a YYYY-MM-DD date")


def dukas_pairs(cfg, only=None):
    """The instruments to take from Dukascopy, by the name they use for
    them. No name, no download: that is how the source is turned off."""
    names = listed(cfg, "dukascopy.pairs", [])
    if only is None:
        return names
    wanted = list(dict.fromkeys(only))
    missing = [n for n in wanted if n not in names]
    if missing:
        raise Fatal(f"not in dukascopy.pairs: {', '.join(missing)}")
    return wanted


def dbento_symbols(cfg, only=None):
    """The instruments to buy from Databento, by the symbol their dataset
    knows them as. No symbol, no download: that is how the source is turned
    off, as it is for Dukascopy."""
    names = listed(cfg, "databento.symbols", [])
    if only is None:
        return names
    wanted = list(dict.fromkeys(only))
    missing = [n for n in wanted if n not in names]
    if missing:
        raise Fatal(f"not in databento.symbols: {', '.join(missing)}")
    return wanted


def dbento_shape(cfg, symbol):
    """What kind of name this is, which the name itself says: `GC.v.0` is one
    continuous series, `GC.FUT` is every contract of a product.

    Read off the symbol rather than configured, because `databento.stype` is
    one setting for the whole section and these two live side by side: the
    continuous series of gold is worth keeping for the years the contracts
    are not bought for. A shape this does not know falls back on the setting
    where there is one, and is refused by name where there is not - asking
    with the wrong stype buys the wrong thing rather than failing.
    """
    parts = symbol.rsplit(".", 2)
    if len(parts) == 3 and parts[1] in CONTINUOUS_RULES and parts[2].isdigit():
        return "continuous"
    if symbol.endswith(".FUT"):
        return "parent"
    given = value(cfg, "databento.stype")
    if given:
        return given
    raise Fatal(f"databento: {symbol} is neither a continuous series (GC.v.0) nor a "
                f"product (GC.FUT); set databento.stype if it is a shape they have added")


def contract_name(symbol, when):
    """What a contract is stored as: `GCZ4` fetched in 2024 is `GCZ24`.

    Their symbol carries one digit of the year, so `GCZ4` is December 2014 as
    much as December 2024, and kept as it stands the two would share a pair
    and ten years of prices would interleave. The digit is read against the
    month being fetched: of the expiries ending in it, the first one from that
    month onwards, since a contract has no prices after it expires.

    That is exact as long as nothing is listed ten years ahead, which is what
    it would take for two candidates to be alive at once. Gold reaches five
    years out and the indices less; crude oil reaches nine, and for a product
    listed further than that this would name the wrong decade - the expiry
    would have to be bought from the definition schema instead.
    """
    code, digit = symbol[-2], symbol[-1]
    month = CONTRACT_MONTHS.get(code)
    if month is None or not digit.isdigit():
        raise Fatal(f"databento: {symbol} does not end in a month code and a year digit")
    year = when.year - when.year % 10 + int(digit)
    # The expiry month counts as alive: a contract trades inside it, up to a
    # day of it that nothing here needs to know.
    if (year, month) < (when.year, when.month):
        year += 10
    return f"{symbol[:-1]}{year % 100:02d}"


def dbento_spend(cfg, given=None):
    """What a run may spend, in dollars. It is not a brake on the pace -
    a day of minutes costs three thousandths - but on the date: asking for
    fifteen years by mistake stops here rather than on the invoice."""
    spend = given if given else value(cfg, "databento.spend", DBENTO_SPEND)
    try:
        spend = float(spend)
    except (TypeError, ValueError):
        raise Fatal(f"config: databento.spend is {spend!r}, not an amount of dollars")
    if spend <= 0:
        raise Fatal(f"config: databento.spend is {spend}, which would buy nothing")
    return spend


def dbento_digits(rows):
    """How many decimal places these prices are quoted to, read off them.

    Theirs to tell and they charge for telling: the tick of an instrument is
    in the definition schema, which is bought like any other data. It is
    also written all over the prices themselves - every one of them is a
    whole number of ticks - so the greatest common divisor of a month of
    them is the tick, and the places it needs are the places to show.

    Gold futures move by a tenth, so their prices are all multiples of
    100000000 in ninths of a thousand million: one decimal place. The E-mini
    moves by a quarter, which needs two.

    Counted rather than divided. The common divisor of every price would be
    the tick exactly, and one price off the tick - a bad print, a rounding
    at their end - drags it down to nothing and asks for nine places. So
    each price says how many places it needs, and the answer is the fewest
    that write all but one in a hundred of them: a handful of odd ones
    cannot decide for the month.

    It decides how the page writes the number and nothing else - the prices
    are stored as they arrive - so reading one place too few costs a digit
    on screen and no data at all.
    """
    counted = {}
    total = 0
    for row in rows:
        for name in ("open", "high", "low", "close"):
            places = max(0, -(Decimal(int(row[name])) * DBENTO_SCALE).normalize().as_tuple().exponent)
            counted[places] = counted.get(places, 0) + 1
            total += 1
    if not total:
        return DBENTO_DIGITS
    enough = total * 0.99
    seen = 0
    for places in sorted(counted):
        seen += counted[places]
        if seen >= enough:
            return places
    return max(counted)


def dukas_budget(cfg):
    budget = value(cfg, "dukascopy.budget", DUKAS_BUDGET)
    if not isinstance(budget, int) or budget < 1:
        raise Fatal(f"config: dukascopy.budget is {budget}, not a number of requests")
    return budget


def tick_sides(cfg):
    bought = listed(cfg, "databento.symbols", [])
    from_bought = date_text(cfg, "databento.start")
    databento = "none" if not bought else (
        f"{' '.join(bought)} from {value(cfg, 'databento.dataset', '?')}, "
        f"{f'from {from_bought}' if from_bought else 'no start date'}, "
        f"up to ${dbento_spend(cfg):.2f} a run, "
        f"{'key set' if value(cfg, 'databento.key') else 'NO KEY'}"
    )
    sides = listed(cfg, "ctrader.ticks.sides", [])
    unknown = [s for s in sides if s not in SIDES]
    if unknown:
        raise Fatal(f"config: ticks.sides: {' '.join(unknown)}; choose among {' '.join(SIDES)}")
    return sides


def tick_budget(cfg):
    budget = value(cfg, "ctrader.ticks.budget", TICK_BUDGET)
    if not isinstance(budget, int) or budget < 1:
        raise Fatal(f"config: ticks.budget is {budget!r}, not a number of requests")
    return budget


def environment(cfg):
    env = value(cfg, "ctrader.account.environment", "live")
    if env not in HOSTS:
        raise Fatal(f"config: account.environment is {env!r}, not live or demo")
    return env


def yq(*args, env=None):
    try:
        return subprocess.run(["yq", *args], capture_output=True, text=True, check=True, env=env).stdout
    except FileNotFoundError:
        raise Fatal("yq not found: install yq-go")
    except subprocess.CalledProcessError as e:
        raise Fatal(f"yq: {e.stderr.strip()}")


# What the first layout kept at the top of the file, and where each of them
# lives now. They were named for a time when cTrader was the only source, so
# `pairs` and `start` needed no owner; with three sources they say nothing
# about whose they are.
MOVED = {
    "account": "ctrader.account",
    "pairs": "ctrader.pairs",
    "periods": "ctrader.periods",
    "start": "ctrader.start",
    "ticks": "ctrader.ticks",
}


def check_layout(cfg, path):
    """Stop on a configuration written for the old shape.

    Reading it as it stands would find no pairs and no periods and conclude
    there is nothing to collect - which is the one way for this to go wrong
    quietly, and on a collector that runs from cron nobody would see it for
    a week.
    """
    stale = [key for key in MOVED if key in cfg]
    if not stale:
        return
    moves = "\n".join(f"  {key}: -> {MOVED[key]}:" for key in stale)
    raise Fatal(
        f"{path} has the old layout: what belongs to cTrader now lives under "
        f"ctrader:, since three sources have their own pairs and their own "
        f"start date.\n{moves}\n"
        f"Move them there with their values, indented under ctrader:, and "
        f"delete them from the top of the file."
    )


def load_config(path):
    if not os.path.exists(path):
        raise Fatal(f"{path} not found")
    # An empty file reads as null.
    return json.loads(yq("-o", "json", ".", path)) or {}


def shared(name):
    installed = os.path.join(SHARE, name)
    if os.path.isfile(installed):
        return installed
    return os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])), name)


def create_config(path):
    """Start the file that config is about to fill: a copy of the template,
    comments and all, when one is at hand; an empty file otherwise, which yq
    fills just as well."""
    template = shared("config.yaml")
    if not os.path.isfile(template):
        template = None
    text = ""
    if template:
        with open(template) as f:
            text = f.read()
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    replace_file(path, text, 0o600, (os.getuid(), os.getgid()))
    log(f"created {path}" + (f" from {template}" if template else ", without comments: no template found"))


def write_config(path, values):
    """Set each dotted path of values in the file, through yq so that the
    comments stay. The values travel in the environment rather than on the
    command line, where ps would show the secrets."""
    env = dict(os.environ)
    steps = []
    for i, (key, v) in enumerate(values.items()):
        var = f"OHLCDB_VALUE_{i}"
        if isinstance(v, str):
            env[var] = v
            # A string that yq adds on its own comes out unquoted, and one
            # like 0123 would then read back as a number.
            steps.append(f'.{key} = strenv({var}) | .{key} style="double"')
        else:
            env[var] = json.dumps(v)
            steps.append(f".{key} = env({var})")
    text = yq(" | ".join(steps), path, env=env)
    st = os.stat(path)
    replace_file(path, text, st.st_mode & 0o7777, (st.st_uid, st.st_gid))


def dir_owner(path):
    """The owner of path's directory, which is the user crond runs collect as.

    Run as root from a shell, as config and accounts are, the files written
    next to the tokens must still end up belonging to that user.
    """
    st = os.stat(os.path.dirname(path) or ".")
    return st.st_uid, st.st_gid


def replace_file(path, text, mode, owner):
    """Write text to path through a rename, so a crash leaves the old file or
    the new one, never half of either."""
    tmp = path + ".tmp"
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, mode)
    with os.fdopen(fd, "w") as f:
        # The mode given to os.open goes through the umask.
        os.fchmod(fd, mode)
        # Only root can give a file away, and only root needs to.
        if os.geteuid() == 0:
            os.fchown(fd, *owner)
        f.write(text)
    os.replace(tmp, path)


def lock(token_file):
    # Two runs refreshing the token at once would invalidate each other's
    # result, so the second one steps aside. The descriptor stays open on
    # purpose: the lock lasts as long as the process.
    fd = os.open(token_file + ".lock", os.O_WRONLY | os.O_CREAT, 0o600)
    if os.geteuid() == 0:
        os.fchown(fd, *dir_owner(token_file))
    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        return False
    return True


def read_tokens(token_file):
    try:
        with open(token_file) as f:
            return json.load(f)
    except FileNotFoundError:
        return None


def save_tokens(token_file, tokens):
    # The pair being replaced is kept beside the new one: a chain of tokens
    # a month long should not hang on a single write.
    previous = read_tokens(token_file)
    if previous is not None:
        replace_file(token_file + ".old", json.dumps(previous), 0o600, dir_owner(token_file))
    replace_file(token_file, json.dumps(tokens), 0o600, dir_owner(token_file))


def authorize(client, cfg, token_file, seed=None):
    """Authenticate the application and return a valid access token.

    A seed is a pair straight from the Playground, whose expiry nobody
    knows: it is refreshed at once, which both proves it and dates it. Only
    what comes back is saved, so a pair the server turns down leaves the
    stored one as it was.
    """
    client.request(APPLICATION_AUTH_REQ, {
        "clientId": need(cfg, "ctrader.client_id"),
        "clientSecret": need(cfg, "ctrader.client_secret"),
    })
    tokens = seed or read_tokens(token_file)
    if tokens is None:
        raise NoTokens(f"no tokens in {token_file}")
    if seed is None and tokens["expires_at"] - time.time() > REFRESH_MARGIN:
        return tokens["access_token"]

    # The refresh invalidates the old tokens: failing to save the new ones
    # would lose both, so make sure they can be written before asking.
    token_dir = os.path.dirname(token_file) or "."
    if not os.access(token_dir, os.W_OK):
        raise Fatal(f"{token_dir} is not writable, cannot refresh the token")
    try:
        res = client.request(REFRESH_TOKEN_REQ, {"refreshToken": tokens["refresh_token"]})
    except ApiError as e:
        # A refresh spends its pair: the server hands back a new one and
        # turns down the old. So a pair from the Playground works once, and
        # from then on the only live one is in the token file.
        raise NoTokens(f"{e}; a refreshed pair cannot be used again") from None
    tokens = {
        "access_token": res["accessToken"],
        "refresh_token": res["refreshToken"],
        "expires_at": int(time.time()) + int(res["expiresIn"]),
    }
    save_tokens(token_file, tokens)
    log(f"access token refreshed, expires {datetime.fromtimestamp(tokens['expires_at'], UTC):%Y-%m-%d}")
    return tokens["access_token"]


def ask_pair():
    """A pair from the Playground, asked for and not kept as it is given:
    refreshing it invalidates it, so only what comes back is worth storing."""
    print("Tokens from the Playground, with the accounts scope")
    return {
        "access_token": ask("access token", secret=True, required=True),
        "refresh_token": ask("refresh token", secret=True, required=True),
        "expires_at": 0,
    }


def ensure_token(client, cfg, token_file, new=False):
    """A valid access token, asking for a pair when there is no other way.

    What is stored is tried first, and refreshed when it is close to
    expiring. Only when there is nothing stored, or the server turns it
    down, is a pair asked for: the question is the way back from a broken
    chain, not a step of every run.
    """
    if not new:
        try:
            return authorize(client, cfg, token_file)
        except NoTokens as e:
            log(f"{e}")
    return authorize(client, cfg, token_file, ask_pair())


@contextmanager
def session(cfg):
    # The symbol list alone can exceed the default 1 MiB message limit.
    with connect(f"wss://{HOSTS[environment(cfg)]}:{PORT}", open_timeout=TIMEOUT, max_size=None) as ws:
        yield Client(ws)


def auth_account(client, account, token):
    try:
        client.request(ACCOUNT_AUTH_REQ, {"ctidTraderAccountId": account, "accessToken": token})
    except ApiError as e:
        raise Fatal(f"{e}; check account.id and account.environment against {SELF} ctrader accounts")


@contextmanager
def account_session(cfg, token_file):
    with session(cfg) as client:
        auth_account(client, need(cfg, "ctrader.account.id"), authorize(client, cfg, token_file))
        yield client


def accounts(client, token):
    res = client.request(GET_ACCOUNTS_REQ, {"accessToken": token})
    scope = res.get("permissionScope")
    return SCOPES.get(scope, scope), res.get("ctidTraderAccount", [])


def describe(account):
    kind = "live" if account.get("isLive") else "demo"
    return (f"{account['ctidTraderAccountId']}  {kind}  "
            f"login {account.get('traderLogin', '?')}  {account.get('brokerTitleShort', '')}")


def list_accounts(client, token):
    scope, found = accounts(client, token)
    print(f"scope: {scope}")
    for a in found:
        print(describe(a))


def make_token(cfg, token_file, new):
    """The token command: leave a valid pair in the token file, and say what
    it reaches, which is the only proof the pair is any good."""
    if new and not sys.stdin.isatty():
        raise Fatal(f"{SELF} ctrader token -n asks for a pair and needs a terminal")
    with session(cfg) as client:
        scope, found = accounts(client, ensure_token(client, cfg, token_file, new))
    if not found:
        raise Fatal("the token reaches no account")
    until = datetime.fromtimestamp(read_tokens(token_file)["expires_at"], UTC)
    print(f"{token_file}: valid until {until:%Y-%m-%d}, scope {scope}, "
          f"{len(found)} account{'s' if len(found) != 1 else ''}")


def symbols(client, account):
    """The pairs the account can trade: name to (symbol id, description)."""
    res = client.request(SYMBOLS_LIST_REQ, {"ctidTraderAccountId": account})
    return {
        s["symbolName"]: (int(s["symbolId"]), s.get("description", ""))
        for s in res.get("symbol", []) if s.get("symbolName")
    }


def check_pairs(available, names, hint=""):
    missing = [n for n in names if n not in available]
    if not missing:
        return
    # Brokers often add a suffix to the name, so offer the near matches.
    notes = []
    for n in missing:
        near = sorted(a for a in available if n[:3].upper() in a.upper())
        notes.append(n + (f" (similar: {', '.join(near)})" if near else ""))
    raise Fatal(f"not on this account: {'; '.join(notes)}{hint}")


def say_digits(db, pair, digits):
    """How the page is to write this instrument's prices, kept at the finest
    reading so far. Never downwards: a quiet month whose prices all happen to
    land on a coarser step would otherwise take a decimal away from one that
    had already been seen."""
    db.execute(
        "UPDATE pairs SET digits = greatest(coalesce(digits, 0), %s) WHERE id = %s"
        "  AND coalesce(digits, -1) < %s",
        (digits, pair, digits),
    )


def say_pip(db, pair, at):
    """Where the pip sits for a source with no broker to ask: at the last
    digit of the price, which on gold futures is the tick as well. It is not
    the tick everywhere - a contract moving by a quarter with two decimals
    has a finer last digit than tick - but it is the only answer the prices
    themselves hold, and it keeps a size in pips meaning about the same
    thing from one feed to the next.

    Only where nothing is known yet: cTrader says it properly, and a figure
    corrected by hand is worth more than this guess and stays."""
    db.execute(
        "UPDATE pairs SET pip_position = %s WHERE id = %s AND pip_position IS NULL",
        (at, pair),
    )


def say_feed(db, feed, market, source):
    """What kind of market a feed's prices come from and who handed them
    over, said again in case the feed was made before the columns existed -
    or before this knew. One statement that usually updates nothing, rather
    than a migration for somebody to remember."""
    db.execute(
        "UPDATE feeds SET market = %s, source = %s WHERE id = %s"
        "  AND (market, source) IS DISTINCT FROM (%s, %s)",
        (market, source, feed, market, source),
    )
    return feed


def ensure(db, select, insert, key, extra=()):
    """The id of the row that select finds by key, inserted with key and
    extra when missing. Looking first spares the identity sequence a number
    on every run that finds the row already there."""
    row = db.execute(select, key).fetchone()
    if row is None:
        db.execute(insert + " ON CONFLICT DO NOTHING", key + extra)
        row = db.execute(select, key).fetchone()
    return row[0]


def feed_id(db, cfg):
    broker = ensure(
        db, "SELECT id FROM brokers WHERE name = %s",
        "INSERT INTO brokers (name) VALUES (%s)",
        (need(cfg, "ctrader.account.broker"),),
    )
    feed = ensure(
        db, "SELECT id FROM feeds WHERE broker_id = %s AND environment = %s AND account_type = %s",
        "INSERT INTO feeds (broker_id, environment, account_type, market, source)"
        "  VALUES (%s, %s, %s, %s, %s)",
        (broker, environment(cfg), need(cfg, "ctrader.account.type")), (CTRADER_MARKET, CTRADER_SOURCE),
    )
    return say_feed(db, feed, CTRADER_MARKET, CTRADER_SOURCE)


def describe_symbol(client, account, symbol_id):
    """What the light symbol list leaves out: the digits, where the pip
    sits, and how many units a lot holds. Only a pair that lacks them pays
    for asking."""
    res = client.request(SYMBOL_BY_ID_REQ, {"ctidTraderAccountId": account, "symbolId": [symbol_id]})
    found = res.get("symbol", [])

    def field(name):
        value = next((s[name] for s in found if name in s), None)
        return None if value is None else int(value)

    # Volumes travel in hundredths of a unit, lot sizes among them: gold
    # comes back as 10000 for its hundred ounces.
    lot = field("lotSize")
    return field("digits"), field("pipPosition"), None if lot is None else lot / 100


def pair_id(db, client, account, feed, name, symbol):
    select = "SELECT id, pip_position, lot_size FROM pairs WHERE feed_id = %s AND name = %s"
    row = db.execute(select, (feed, name)).fetchone()
    symbol_id, description = symbol
    if row:
        # A pair stored before these were asked for: one request, once.
        if row[1] is None or row[2] is None:
            _, pip, lot = describe_symbol(client, account, symbol_id)
            if pip is not None or lot is not None:
                db.execute(
                    "UPDATE pairs SET pip_position = coalesce(%s, pip_position),"
                    " lot_size = coalesce(%s, lot_size) WHERE id = %s",
                    (pip, lot, row[0]),
                )
                log(f"{name}: pip position {pip}, lot of {lot} units")
        return row[0]
    digits, pip, lot = describe_symbol(client, account, symbol_id)
    return ensure(
        db, select,
        "INSERT INTO pairs (feed_id, name, description, digits, pip_position, lot_size)"
        " VALUES (%s, %s, %s, %s, %s, %s)",
        (feed, name), (description, digits, pip, lot),
    )


def ms(dt):
    return int(dt.timestamp() * 1000)


def decode(bar):
    # Prices are integers in units of 1e-5 whatever the symbol's digits, and
    # the JSON leaves out fields that are zero: a missing delta means that
    # price equals the low.
    low = int(bar.get("low", 0))
    open_, high, close = (
        Decimal(low + int(bar.get(k, 0))).scaleb(-5)
        for k in ("deltaOpen", "deltaHigh", "deltaClose")
    )
    ts = datetime.fromtimestamp(int(bar["utcTimestampInMinutes"]) * 60, UTC)
    return ts, open_, high, Decimal(low).scaleb(-5), close, int(bar.get("volume", 0))


def fetch(client, account, symbol_id, period, start, end):
    """Return the candles between start and end, oldest first."""
    bars = []
    while start < end:
        time.sleep(HISTORY_INTERVAL)
        res = client.request(GET_TRENDBARS_REQ, {
            "ctidTraderAccountId": account,
            "symbolId": symbol_id,
            "period": PERIODS[period][0],
            "fromTimestamp": ms(start),
            # The server includes a candle opening at toTimestamp, which the
            # next request would then fetch again as its first.
            "toTimestamp": ms(end) - 1,
        })
        got = sorted(decode(b) for b in res.get("trendbar", []))
        bars[:0] = got
        # A capped reply keeps the newest candles and sets hasMore: ask again
        # for the part before the first one received.
        if not res.get("hasMore") or not got or got[0][0] >= end:
            break
        end = got[0][0]
    return bars


def store(db, pair, period, bars):
    if bars:
        with db.cursor() as cur:
            cur.executemany(UPSERT, [(pair, period, *bar) for bar in bars])
    # Commit per request, so an interrupted backfill keeps what it already got.
    db.commit()


def collect_period(client, db, account, pair, symbol_id, name, period, start, stop):
    span = min(timedelta(minutes=PERIODS[period][1] * CANDLES_PER_REQUEST), MAX_SPAN)
    began = time.monotonic()
    received, last, t = 0, None, start
    while t < stop:
        end = min(t + span, stop)
        bars = fetch(client, account, symbol_id, period, t, end)
        store(db, pair, period, bars)
        if bars:
            received += len(bars)
            last = bars[-1][0]
        # A backfill takes a while. A terminal gets one line that keeps
        # itself up to date; a log gets a line per month instead, which the
        # escapes of the other would clutter.
        if TTY:
            done = (end - start) / (stop - start)
            elapsed = time.monotonic() - began
            progress(f"{name}: {done:.0%}, {received} candles up to {end:%Y-%m-%d}, "
                     f"{duration(elapsed / done - elapsed)} left")
        elif end.month != t.month:
            log(f"{name}: {received} candles up to {end:%Y-%m-%d}")
        t = end
    if received:
        log(f"{name}: {received} candles received, last {last:%Y-%m-%d %H:%M} UTC")
    else:
        log(f"{name}: no candles from {start:%Y-%m-%d %H:%M} to {stop:%Y-%m-%d %H:%M} UTC")


def fetch_ticks(client, account, symbol_id, side, start, end):
    """The ticks of a window, oldest first, and whether the reply was capped.

    A capped reply holds the newest ticks of the window and leaves the older
    part of it unanswered, so the window has to be asked again in halves.
    """
    time.sleep(HISTORY_INTERVAL)
    res = client.request(GET_TICKDATA_REQ, {
        "ctidTraderAccountId": account,
        "symbolId": symbol_id,
        "type": SIDES[side],
        "fromTimestamp": ms(start),
        "toTimestamp": ms(end) - 1,
    })
    # Newest first, and only the first tick is whole: the rest are
    # differences from the one before, running backwards.
    when = price = 0
    ticks = []
    for i, tick in enumerate(res.get("tickData", [])):
        step_time, step_price = int(tick["timestamp"]), int(tick["tick"])
        when = step_time if i == 0 else when + step_time
        price = step_price if i == 0 else price + step_price
        ticks.append((datetime.fromtimestamp(when / 1000, UTC), Decimal(price).scaleb(-5)))
    ticks.reverse()
    return ticks, bool(res.get("hasMore"))


def store_ticks(db, pair, side, ticks, months):
    """Writes a window's ticks, making a month's partition the first time
    anything lands in it."""
    if not ticks:
        return
    for month in {(when.year, when.month) for when, _ in ticks} - months:
        db.execute("SELECT tick_partition(%s)", (datetime(*month, 1, tzinfo=UTC).date(),))
        months.add(month)
    with db.cursor() as cur:
        cur.executemany(TICK_INSERT, [(pair, side, when, price) for when, price in ticks])
    db.commit()


def collect_ticks(client, db, account, pair, symbol_id, name, sides, first, budget, stop=None):
    """The ticks of a pair, from where the last run left off, for as many
    requests as the budget allows; None as a budget means no limit. Returns
    what is left of it."""
    stop = stop or datetime.now(UTC)
    months = set()
    requests = 0
    for side in sides:
        last = db.execute(
            "SELECT max(ts) FROM ticks WHERE pair_id = %s AND side = %s", (pair, side)
        ).fetchone()[0]
        cursor = last + timedelta(milliseconds=1) if last else first
        if cursor is None:
            log(f"{name} {side} ticks: nothing stored yet, skipped; ticks.start in the config starts it")
            continue
        window = FIRST_TICK_WINDOW
        received = 0
        while cursor < stop and (budget is None or budget > 0):
            end = min(cursor + window, stop)
            ticks, capped = fetch_ticks(client, account, symbol_id, side, cursor, end)
            requests += 1
            budget = budget if budget is None else budget - 1
            if capped and end - cursor > MIN_TICK_WINDOW:
                window = max((end - cursor) / 2, MIN_TICK_WINDOW)
                continue
            store_ticks(db, pair, side, ticks, months)
            received += len(ticks)
            cursor = end
            # A window that came back nearly empty can afford to be wider.
            if len(ticks) * 3 < TICKS_PER_REPLY and window < MAX_TICK_WINDOW:
                window = min(window * 2, MAX_TICK_WINDOW)
            if TTY:
                progress(f"{name} {side} ticks: {received} up to {cursor:%Y-%m-%d %H:%M}, "
                         f"{requests} requests")
        log(f"{name} {side} ticks: {received} stored, up to {cursor:%Y-%m-%d %H:%M} UTC"
            + ("" if cursor >= stop else ", more next run"))
    return budget


def roll_pair(db, pair, what, wanted, since):
    """The longer candles of one pair, one length at a time.

    Each is a single statement, so nothing can be said while it runs: on
    twenty years of minutes the first one takes a while, and a line before
    it is the only warning there is that the program has not stopped.
    """
    built = []
    for period in wanted:
        if TTY:
            progress(f"{what}: building {period} from the minutes")
        written = db.execute("SELECT roll_bars(%s, %s, %s)", (pair, period, since)).fetchone()[0]
        # Per length, so that an interrupted first run keeps what it built.
        db.commit()
        if written:
            built.append(f"{written} {period}")
    if built:
        log(f"{what}: built {', '.join(built)}")


def roll_all(db, cfg, only=None, since=None, lengths=None):
    """The longer candles of every configured pair, where its source does
    not publish them: what cTrader is asked for is downloaded and left
    alone, and the rest is made from the minutes. A candle from the broker
    always beats one worked out here.

    Nothing is rebuilt from the start unless a date says so: each length
    carries on from the last one built, its own candle made again in case
    the period was still open when it was.

    The futures contracts of a product are left with their minutes and
    nothing else, and that needs no rule here: the pairs are looked up by
    the names the configuration holds, and a contract is never one of them -
    what is configured is the product. Building the hours of two hundred
    contracts to draw none of them is the thing this quietly avoids.
    """
    wanted = only_of(rollup(cfg), lengths)
    if not wanted:
        return
    known_pairs(cfg, only)
    sources = []
    theirs = only_of(dukas_pairs(cfg), only)
    if theirs:
        sources.append((dukas_feed(db), theirs, {"M1"}, " dukascopy"))
    bought = only_of(dbento_symbols(cfg), only)
    if bought:
        sources.append((dbento_feed(db, cfg), bought, {"M1"}, " databento"))
    ours = only_of(listed(cfg, "ctrader.pairs", []), only)
    if ours:
        sources.append((feed_id(db, cfg), ours, set(periods(cfg)), ""))
    db.commit()
    for feed, names, published, tag in sources:
        for name in names:
            row = db.execute(
                "SELECT id FROM pairs WHERE feed_id = %s AND name = %s", (feed, name)
            ).fetchone()
            # A pair named in the configuration but never collected has
            # nothing to build from yet.
            if row is None:
                continue
            roll_pair(db, row[0], f"{name}{tag}", [p for p in wanted if p not in published], since)


def roll_only(cfg, since, only, lengths=None):
    """The building and nothing else, for a first run over a history that
    was collected before there was anything to build it with."""
    known_periods(cfg, lengths)
    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        roll_all(db, cfg, only, since, lengths)
        # And the stitched series of every futures product, from the start:
        # this is the command for a history collected before there was
        # anything to build from it, and the pieces are part of that. A
        # collection does its own product and only the days it bought, which
        # is what keeps a run every five minutes cheap.
        pieces = db.execute("SELECT rebuild_roll_steps()").fetchone()[0]
        db.commit()
        if pieces:
            log(f"{pieces} pieces of the stitched series")


def only_of(configured, only):
    """The names of one source among the ones asked for.

    A pair named with -p belongs to one source and not to the others, so
    where several are collected together each takes the share it knows and
    leaves the rest alone. The single-source commands keep their own
    checking, which is stricter on purpose: asking `dukascopy -p GC.v.0` is
    a mistake worth being told about, where `collect -p GC.v.0` is not.
    """
    if only is None:
        return configured
    wanted = set(only)
    return [name for name in configured if name in wanted]


def known_pairs(cfg, only):
    """Every name any source is configured for, so that a -p nobody knows is
    caught once rather than by whichever source is asked first."""
    if only is None:
        return
    known = set(listed(cfg, "ctrader.pairs", [])) | set(listed(cfg, "dukascopy.pairs", [])) \
        | set(listed(cfg, "databento.symbols", []))
    missing = [n for n in dict.fromkeys(only) if n not in known]
    if missing:
        raise Fatal(f"not configured anywhere: {' '.join(missing)}")


def known_periods(cfg, lengths):
    """The same for -t: a length is either one the broker is asked for or
    one built from the minutes. Anything else is refused rather than
    fetched, so that -t only ever narrows what the configuration says."""
    if lengths is None:
        return
    fetched = periods(cfg)
    built = rollup(cfg)
    missing = [p for p in dict.fromkeys(lengths) if p not in fetched and p not in built]
    if missing:
        raise Fatal(f"not configured: {' '.join(missing)}; ctrader.periods fetches "
                    f"{' '.join(fetched)} and rollup builds {' '.join(built) or 'nothing'}")


def chosen_pairs(cfg, only):
    """The configured pairs, or the ones asked for among them."""
    configured = listed(cfg, "ctrader.pairs", [])
    if not configured:
        raise Fatal(f"no pairs configured: {SELF} pair add adds them")
    names = list(dict.fromkeys(only)) if only else configured
    unknown = [n for n in names if n not in configured]
    if unknown:
        raise Fatal(f"not in the configuration: {' '.join(unknown)}; {SELF} pair add adds them")
    return names


def collect_ctrader(db, cfg, token_file, since, until, only, budget, lengths=None):
    """The account's own candles, and then the share of the ticks the budget
    allows. Nothing here connects while no pair of the account is configured,
    so an install that uses only the free sources needs no cTrader account at
    all. True when anything was stored."""
    names = only_of(listed(cfg, "ctrader.pairs", []), only)
    if not names:
        return False
    chosen = only_of(periods(cfg), lengths)
    first = start_date(cfg)
    # A tick is not a length of candle, and a run asked for one length is
    # filling that one: the ticks wait for a run that did not choose.
    sides = tick_sides(cfg) if lengths is None else []
    if not chosen and not sides:
        return False
    ticks_first = start_date(cfg, "ctrader.ticks.start")
    ran = False
    account = need(cfg, "ctrader.account.id")
    feed = feed_id(db, cfg)
    db.commit()
    with account_session(cfg, token_file) as client:
        available = symbols(client, account)
        check_pairs(available, names, f"; {SELF} pair search looks them up")
        for name in names:
            pair = pair_id(db, client, account, feed, name, available[name])
            db.commit()
            for period in chosen:
                last = db.execute(
                    "SELECT max(ts) FROM bars WHERE pair_id = %s AND period = %s", (pair, period)
                ).fetchone()[0]
                # Just after the last stored candle, which is final: the
                # server hands out closed candles only. A millisecond
                # rather than a period, as a month has no fixed length.
                start = since or (last + timedelta(milliseconds=1) if last else None)
                if start is None and first:
                    log(f"{name} {period}: nothing stored yet, starting from {first:%Y-%m-%d}")
                    start = first
                if start is None:
                    # A pair or period added later must not stop the others.
                    log(f"{name} {period}: nothing stored yet, skipped; "
                        f"{SELF} -s DATE -p {name}, or start in the config, starts it")
                    continue
                now = datetime.now(UTC)
                collect_period(client, db, account, pair, available[name][0], f"{name} {period}",
                               period, start.astimezone(UTC), min(until, now) if until else now)
                ran = True
            # After the candles, so that a long catch-up of ticks never
            # holds them up.
            if sides:
                budget = collect_ticks(client, db, account, pair, available[name][0], name,
                                       sides, ticks_first, budget)
                ran = True
    return ran


def collect(cfg, token_file, since, until, only, lengths=None):
    """Everything configured: the history published by Dukascopy, what was
    bought from Databento, then the account's own candles and ticks. Each
    source is off when it has no instruments named in it."""
    known_pairs(cfg, only)
    known_periods(cfg, lengths)
    names = only_of(listed(cfg, "ctrader.pairs", []), only)
    theirs = only_of(dukas_pairs(cfg), only)
    bought = only_of(dbento_symbols(cfg), only)
    if not names and not theirs and not bought:
        raise Fatal(f"no pairs configured: {SELF} pair add adds them, dukascopy.pairs "
                    f"names what to take from their history, and databento.symbols "
                    f"what to buy")
    # The free history and what was bought are minutes and nothing else, so
    # a -t without M1 leaves them to be built from rather than fetched; and
    # a run that was asked only for lengths built here fetches nothing on
    # purpose, which is not the missing start date the end complains of.
    minutes = bool(only_of(["M1"], lengths))
    fetching = (minutes and bool(theirs or bought)) or bool(names and only_of(periods(cfg), lengths))
    ran = False
    # A source that fails is that source's failure, not the run's: the free
    # history should not be held back by a bill, and what did arrive should
    # get its longer candles either way. Each is kept to itself here, and
    # what went wrong is said again at the end so that the run is not
    # reported as a success. Only Fatal, which is a refusal we understand -
    # a database that has gone is not something to carry on through.
    failed = []

    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        if theirs and minutes:
            try:
                collect_dukascopy(db, cfg, since, until, only, dukas_budget(cfg))
                ran = True
            except Fatal as e:
                # Whatever was in flight goes; every day and month of these
                # commits as it lands, so this drops the unfinished one.
                db.rollback()
                log(str(e))
                failed.append("dukascopy")
        if bought and minutes:
            try:
                collect_databento(db, cfg, since, until, only, dbento_spend(cfg))
                ran = True
            except Fatal as e:
                db.rollback()
                log(str(e))
                failed.append("databento")
        try:
            if collect_ctrader(db, cfg, token_file, since, until, only, tick_budget(cfg), lengths):
                ran = True
        except Fatal as e:
            db.rollback()
            log(str(e))
            failed.append("ctrader")
        # Last of all, over everything stored by this run and by the ones
        # before it: the lengths nobody publishes are worked out from the
        # minutes that just arrived.
        roll_all(db, cfg, only, since, lengths)
    if failed:
        raise Fatal(f"{' and '.join(failed)} did not finish; the rest of the run did")
    if not ran and fetching:
        raise Fatal(f"nothing stored yet: run {SELF} -s DATE, or set start in the config, "
                    f"to say where the history starts")


def collect_only_ctrader(cfg, token_file, since, until, only, lengths=None):
    """The account's feed and nothing else, as a full collection does for it:
    the candles, then the share of the ticks the configured budget allows.
    `ctrader ticks` is the one that spends as many requests as it takes."""
    if not listed(cfg, "ctrader.pairs", []):
        raise Fatal("config: ctrader.pairs names no instrument, so there is nothing to collect")
    known_periods(cfg, lengths)
    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        collect_ctrader(db, cfg, token_file, since, until, only, tick_budget(cfg), lengths)
        roll_all(db, cfg, only, since, lengths)


def collect_only_ticks(cfg, token_file, since, until, only, budget):
    """The ticks and nothing else, with no budget unless one is given: the
    way to catch up on months of them in one run instead of a share per run.

    The lock is held throughout, so the periodic runs step aside and the
    candles wait for it to finish.
    """
    account = need(cfg, "ctrader.account.id")
    names = chosen_pairs(cfg, only)
    sides = tick_sides(cfg)
    if not sides:
        raise Fatal("config: ticks.sides names no side, so there are no ticks to collect")
    first = since or start_date(cfg, "ctrader.ticks.start")
    now = datetime.now(UTC)
    stop = min(until, now) if until else now

    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        feed = feed_id(db, cfg)
        db.commit()
        with account_session(cfg, token_file) as client:
            available = symbols(client, account)
            check_pairs(available, names, f"; {SELF} pair search looks them up")
            for name in names:
                pair = pair_id(db, client, account, feed, name, available[name])
                db.commit()
                budget = collect_ticks(client, db, account, pair, available[name][0], name,
                                       sides, first, budget, stop)


# -- Dukascopy ---------------------------------------------------------------


def dukas_feed(db):
    """Their history is a feed like any other, so the sessions, the levels
    and the page treat it as one without knowing where it came from."""
    broker = ensure(
        db, "SELECT id FROM brokers WHERE name = %s",
        "INSERT INTO brokers (name) VALUES (%s)",
        (DUKAS_BROKER,),
    )
    feed = ensure(
        db, "SELECT id FROM feeds WHERE broker_id = %s AND environment = %s AND account_type = %s",
        "INSERT INTO feeds (broker_id, environment, account_type, market, source)"
        "  VALUES (%s, %s, %s, %s, %s)",
        (broker, DUKAS_ENVIRONMENT, DUKAS_ACCOUNT), (DUKAS_MARKET, DUKAS_BROKER),
    )
    return say_feed(db, feed, DUKAS_MARKET, DUKAS_BROKER)


def dukas_pair_id(db, feed, name, digits):
    return ensure(
        db, "SELECT id FROM pairs WHERE feed_id = %s AND name = %s",
        "INSERT INTO pairs (feed_id, name, description, digits) VALUES (%s, %s, %s, %s)",
        (feed, name), (f"{DUKAS_BROKER} {DUKAS_ACCOUNT}", digits),
    )


def dukas_stored_digits(db, feed, name):
    """The scale the pair's candles were stored at, where there are any.
    What it is is settled once and kept: a feed at two scales is worse than
    a feed at the wrong one, which at least can be fetched again."""
    row = db.execute(
        "SELECT digits FROM pairs WHERE feed_id = %s AND name = %s", (feed, name)
    ).fetchone()
    return row[0] if row else None


def jetta_fetch(url, what):
    """What their API answers, decoded, or None where it has nothing to
    say. None is not the end of the run: the caller asks the files instead,
    so a change at their end costs speed and not the history."""
    request = urllib.request.Request(url, headers={"User-Agent": DUKAS_AGENT})
    for attempt in range(JETTA_RETRIES):
        refused = False
        try:
            with urllib.request.urlopen(request, timeout=JETTA_TIMEOUT) as res:
                answer = json.loads(res.read())
            return answer if isinstance(answer, dict) else None
        except urllib.error.HTTPError as e:
            # A day they have nothing for is a 400, and they mean it:
            # asking again changes nothing except the time it takes.
            if e.code < 500 and e.code != 429:
                return None
            refused = e.code == 429
            trouble = f"HTTP {e.code}"
        except (urllib.error.URLError, TimeoutError) as e:
            trouble = getattr(e, "reason", e)
        except ValueError as e:
            # An answer that is not the JSON it should be. Their doing, and
            # not one more request will fix it.
            log(f"dukascopy: the API answered {what} with something else ({e})")
            return None
        if attempt == JETTA_RETRIES - 1:
            log(f"dukascopy: the API would not say {what} ({trouble}), asking the files instead")
            return None
        time.sleep(JETTA_COOLDOWN if refused else JETTA_PAUSE)
    return None


def jetta_codes(names):
    """Their code and price scale for each instrument named, by their own
    list rather than by guesswork: the names take every shape they have,
    from XAU-USD to 0005.HK-HKD, and only the hyphens tell them apart.
    The dot goes too: an index is USATECH.IDX-USD to the API and
    USATECHIDXUSD in the path of the files, and the configuration has to be
    the name the files know, since they are the reserve.

    An instrument they do not list, or a list that does not arrive, is left
    out: the caller falls back to the files for it and takes the scale from
    what is already stored.
    """
    listing = jetta_fetch(JETTA_INSTRUMENTS, "which instruments it has")
    if not listing:
        return {}
    def bare(s):
        return s.upper().replace("-", "").replace("/", "").replace(".", "")

    wanted = {bare(n): n for n in names}
    found = {}
    for item in listing.get("instruments", []):
        code = item.get("code") or ""
        name = wanted.get(bare(code))
        scale = item.get("priceScale")
        if name and isinstance(scale, int) and 0 <= scale <= 10:
            found[name] = (code, scale)
    return found


def jetta_history(code, name):
    """The first minute they have of an instrument, so that a pair with no
    start date of its own begins where the history does instead of at a
    date someone had to look up."""
    about = jetta_fetch(JETTA_INSTRUMENT.format(code=code), f"when {name} begins")
    for period in (about or {}).get("histories", []):
        if period.get("period") == "MINUTE" and period.get("from"):
            return datetime.fromtimestamp(period["from"] / 1000, UTC)
    return None


def jetta_candles(answer, digits, what):
    """The minutes of a day, as rows ready to store.

    They send one starting point and then the steps from it: a timestamp
    for the day, the first minute's four prices, and for every minute after
    that how far it moved from the one before - times in units of shift,
    prices in units of multiplier. Volumes are sent whole. Rebuilt, it is
    the same day the files hold, to the last decimal place; the files only
    add the hour the market is shut as flat candles with nothing traded,
    which are left out of both.

    A day the market was shut comes back empty, which is the truth about
    it; an answer in a shape this does not read comes back as None, which
    sends the caller to the files.
    """
    wanted = ("timestamp", "multiplier", "shift", "times", "volumes",
              "open", "high", "low", "close", "opens", "highs", "lows", "closes")
    if not all(key in answer for key in wanted):
        log(f"dukascopy: the API answered {what} in a shape this does not read, "
            f"asking the files instead")
        return None
    times = answer["times"]
    if not times:
        return []
    step = Decimal(str(answer["multiplier"]))
    if step != Decimal(10) ** -digits:
        # Their scale against ours. Storing the difference would put the
        # prices out by a factor of ten and nothing downstream would know.
        raise Fatal(f"dukascopy: {what} is quoted in steps of {step}, not {Decimal(10) ** -digits}; "
                    f"the pair is stored with {digits} decimal places and the two cannot both be right")
    shift = int(answer["shift"]) // 1000
    began = datetime.fromtimestamp(int(answer["timestamp"]) / 1000, UTC)
    sides = ("opens", "highs", "lows", "closes")
    price = {side: Decimal(str(answer[side[:-1]])) for side in sides}
    rows, minutes = [], 0
    for i, gap in enumerate(times):
        minutes += gap
        for side in sides:
            price[side] += Decimal(answer[side][i]) * step
        rows.append((
            began + timedelta(seconds=minutes * shift),
            *(price[side].quantize(Decimal("0.00001")) for side in sides),
            Decimal(str(round(answer["volumes"][i], 5))),
        ))
    return rows


class Pace:
    """How often their server is asked, which is their call and not ours.

    Every refusal makes the run slower for good: the alternative is to keep
    asking at a pace they have already turned down, which ends the run
    instead of finishing it.
    """

    def __init__(self, interval=DUKAS_INTERVAL):
        self.interval = interval
        self.refusals = 0
        self.spent = 0.0

    def wait(self):
        # The interval is how often they are asked, not how long we sit
        # doing nothing: a reply that took them a minute has already left
        # them all the room the interval was meant to leave.
        time.sleep(max(0.0, self.interval - self.spent))
        self.spent = 0.0

    def refused(self, what, after=None):
        self.refusals += 1
        was = self.interval
        self.interval = min(self.interval * 1.5, DUKAS_SLOWEST)
        # Their own answer when they give one, and a minute more for each
        # refusal in a row when they do not - up to five, since a quarter of
        # an hour spent on one file is a quarter of an hour not spent on the
        # rest, and what is stored is stored.
        wait = after if after else min(DUKAS_COOLDOWN * self.refusals, DUKAS_LONGEST_WAIT)
        pace = f", now one file every {self.interval:.1f}s" if self.interval > was else ""
        # Said out loud: a wait nobody is told about looks like a program
        # that has stopped.
        log(f"dukascopy: turned away, waiting {duration(wait)} before asking for {what} again{pace}")
        time.sleep(wait)


def dukas_fetch(url, pace, what):
    """The day's file, or None where there is none: a day the market was
    shut has no file, and neither has one that is not published yet."""
    request = urllib.request.Request(url, headers={"User-Agent": DUKAS_AGENT})
    for attempt in range(DUKAS_RETRIES):
        pace.wait()
        began = time.monotonic()
        try:
            with urllib.request.urlopen(request, timeout=DUKAS_TIMEOUT) as res:
                body = res.read()
            pace.spent = time.monotonic() - began
            pace.refusals = 0
            return body
        except urllib.error.HTTPError as e:
            pace.spent = time.monotonic() - began
            if e.code == 404:
                pace.refusals = 0
                return None
            if e.code not in DUKAS_AGAIN or attempt == DUKAS_RETRIES - 1:
                raise Fatal(f"dukascopy: {url}: HTTP {e.code} {e.reason}"
                            + ("; they are turning us away, so leave them be for a while and "
                               f"run {SELF} dukascopy again - it carries on where it stopped"
                               if e.code in DUKAS_REFUSALS else ""))
            retry_after = e.headers.get("Retry-After") if e.headers else None
            pace.refused(what, int(retry_after) if retry_after and retry_after.isdigit() else None)
        except urllib.error.URLError as e:
            # Not an answer at all: the network, or a wait so long that even
            # this gave up. Worth another go, but it is not a refusal and
            # does not say anything about how fast we may ask.
            if attempt == DUKAS_RETRIES - 1:
                raise Fatal(f"dukascopy: {url}: {e.reason}")
            log(f"dukascopy: no answer for {what} ({e.reason}), trying again")
            time.sleep(DUKAS_COOLDOWN)
    return None


def dukas_candles(raw, day, digits):
    """The minutes of a day, as rows ready to store.

    The file is plain LZMA, the shape the format had before xz put a
    container round it, so it is decompressed alone. Each record is a whole
    minute; an empty minute is written as zeroes and has no candle, which is
    the same rule the rest of the collector follows.
    """
    body = lzma.LZMADecompressor(format=lzma.FORMAT_ALONE).decompress(raw)
    size = struct.calcsize(DUKAS_RECORD)
    scale = Decimal(10) ** -digits
    midnight = datetime.combine(day, datetime.min.time(), UTC)
    rows = []
    for offset in range(0, len(body) - size + 1, size):
        seconds, open_, close, low, high, volume = struct.unpack_from(DUKAS_RECORD, body, offset)
        if volume <= 0 and open_ == close == low == high:
            continue
        price = lambda points: (Decimal(points) * scale).quantize(Decimal("0.00001"))
        rows.append((
            midnight + timedelta(seconds=seconds),
            price(open_), price(high), price(low), price(close),
            Decimal(str(round(volume, 5))),
        ))
    return rows


def collect_dukascopy(db, cfg, since, until, only, budget):
    """Their M1 candles, a day and a request at a time.

    Where the collector left off is where the last stored minute is, rounded
    back to its day so that a day half published is asked for again. A day
    the market was shut holds nothing, and that is not stored and not
    remembered: asking again costs one request, and that is cheaper than a
    table of absences.
    """
    names = dukas_pairs(cfg, only)
    if not names:
        return budget
    first = since or start_date(cfg, "dukascopy.start")
    # Their day is published once it is over, so today is never complete.
    last_day = (min(until, datetime.now(UTC)) if until else datetime.now(UTC)).date() - timedelta(days=1)
    feed = dukas_feed(db)
    db.commit()
    pace = Pace()
    codes = jetta_codes(names)

    for name in names:
        code, scale = codes.get(name, (None, None))
        # What is already stored settles the scale: the candles of a pair
        # are all read the same way or none of them mean anything. A new
        # pair takes theirs, and a pair that is neither cannot be fetched.
        digits = dukas_stored_digits(db, feed, name)
        if digits is None:
            digits = scale
        elif scale is not None and scale != digits:
            log(f"{name} dukascopy: stored with {digits} decimal places, but they now quote it "
                f"to {scale}; keeping {digits} - to change it, delete the pair's candles and "
                f"fetch them again")
        if digits is None:
            log(f"{name} dukascopy: they do not list it and nothing of it is stored, so there is "
                f"no telling what its prices are quoted to; check the name against their own")
            continue
        pair = dukas_pair_id(db, feed, name, digits)
        db.commit()
        stored = db.execute(
            "SELECT max(ts) FROM bars WHERE pair_id = %s AND period = 'M1'", (pair,)
        ).fetchone()[0]
        if since:
            day = since.date()
        elif stored:
            day = stored.date()
        else:
            # Nothing stored and no date given: where their own history
            # starts, which they will say, so that a first run needs no
            # date looked up anywhere.
            began = first or (jetta_history(code, name) if code else None)
            if began is None:
                log(f"{name} dukascopy: no start date and they would not say where it begins; "
                    f"set dukascopy.start or run {SELF} dukascopy -s DATE")
                continue
            day = began.date()
        received = days = files = missed = 0
        while day <= last_day and (budget is None or budget > 0):
            what = f"{name} {day}"
            if budget is not None:
                budget -= 1
            rows = None
            if code:
                answer = jetta_fetch(JETTA_CANDLES.format(
                    code=code, year=day.year, month=day.month, day=day.day), what)
                if answer is not None:
                    rows = jetta_candles(answer, digits, what)
                missed = 0 if rows is not None else missed + 1
                if missed >= JETTA_GIVE_UP:
                    log(f"{name} dukascopy: the API has not answered for {missed} days running, "
                        f"so the rest of this run comes from the files")
                    code = None
            if rows is None:
                # The API had nothing for this day. The archive still does,
                # slowly: a run that falls back for every day is the old
                # one, and finishes all the same.
                raw = dukas_fetch(DUKAS_URL.format(
                    symbol=name, year=day.year, month=day.month - 1, day=day.day), pace, what)
                rows = dukas_candles(raw, day, digits) if raw else []
                files += 1
            if rows:
                store(db, pair, "M1", rows)
                received += len(rows)
                days += 1
                if days == 1:
                    # The first candle of a run says whether the prices came
                    # out at the right scale, which is the one thing here
                    # that no error can report by itself.
                    log(f"{name} dukascopy: {day} opens at {rows[0][1]}")
            if TTY:
                # A run with a budget counts down to it; one without says
                # how far it has to go instead, which is the number the
                # reader of a long catch-up actually wants.
                left = f"{budget} days left in the budget" if budget is not None \
                    else f"{(last_day - day).days} days to go"
                slow = f", {files} from the files, one every {pace.interval:.1f}s" if files else ""
                progress(f"{name} dukascopy: {day}, {received} candles, {left}{slow}")
            day += timedelta(days=1)
        if received:
            log(f"{name} dukascopy: {received} candles over {days} days, up to {day - timedelta(days=1)}"
                + (f", {files} days from the files" if files else ""))
        elif budget is not None and budget <= 0:
            log(f"{name} dukascopy: nothing yet, more next run")
        else:
            log(f"{name} dukascopy: nothing new")
    return budget


def dbento_ask(cfg, method, params, get=False):
    """One call to their historical API.

    Basic auth with the key as the user and no password, which is how they
    take it. The key never reaches the command line - this is a library
    call, not psql - so there is nothing for a process list to leak.

    Most of their methods take the arguments in the body, and one does not:
    metadata.get_dataset_range answers 405 to anything but a GET. Hence the
    flag, rather than a guess that holds for three endpoints out of four.
    """
    key = need(cfg, "databento.key")
    query = urllib.parse.urlencode(params, doseq=True)
    request = urllib.request.Request(
        f"{DBENTO_HOST}/{method}" + (f"?{query}" if get else ""),
        data=None if get else query.encode(),
        headers={
            "Authorization": "Basic " + base64.b64encode(f"{key}:".encode()).decode(),
            "Accept-Encoding": "identity",
        },
    )
    for attempt in range(DBENTO_RETRIES):
        try:
            return urllib.request.urlopen(request, timeout=DBENTO_TIMEOUT)
        except urllib.error.HTTPError as e:
            body = e.read(500).decode(errors="replace").strip()
            # Their own answer, and a refusal they mean: a key that is not a
            # key, a symbol that is not theirs, a date before the dataset.
            if e.code < 500 and e.code != 429:
                raise Fatal(f"databento: {method}: HTTP {e.code} {body}")
            trouble = f"HTTP {e.code} {body}"
        except (urllib.error.URLError, TimeoutError) as e:
            trouble = str(getattr(e, "reason", e))
        if attempt == DBENTO_RETRIES - 1:
            raise Fatal(f"databento: {method}: {trouble}")
        log(f"databento: {method} did not answer ({trouble}), trying again")
        time.sleep(DBENTO_PAUSE)
    raise Fatal(f"databento: {method}: no answer")


def dbento_cost(cfg, params):
    """What that request would cost, in dollars. Free to ask, and asked
    before every download."""
    with dbento_ask(cfg, "metadata.get_cost", params) as res:
        answer = res.read().decode().strip()
    try:
        return float(answer)
    except ValueError:
        raise Fatal(f"databento: metadata.get_cost answered {answer[:80]!r}, not an amount")


def dbento_ready(cfg, dataset, schema):
    """How far their history goes at this moment, for the schema being
    bought, floored to the minute.

    Free to ask, as the cost is, and it moves through the day: what they
    hold is not yesterday's close but a few hours ago, so a run that stops
    at midnight throws away everything since. The end they give is
    exclusive and carries nanoseconds - 07:01:12.407956000 - and the whole
    minute before it is the last one certain to be complete, which is what
    flooring leaves.

    None where they do not answer or answer in a shape this does not read:
    a metadata call that fails is no reason to fail a collection, and the
    caller falls back on the midnight it used before.
    """
    try:
        with dbento_ask(cfg, "metadata.get_dataset_range", {"dataset": dataset}, get=True) as res:
            answer = json.load(res)
        # The schema's own end rather than the dataset's: they agree on the
        # minutes today, and ohlcv-1d shows they need not.
        end = (answer.get("schema", {}).get(schema) or answer).get("end")
        when = datetime.strptime(end[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=UTC)
    except (Fatal, ValueError, TypeError, AttributeError, json.JSONDecodeError) as e:
        log(f"databento: cannot tell how far {dataset} goes ({e}), buying up to midnight")
        return None
    return when.replace(second=0, microsecond=0)


def dbento_contracts(cfg, dataset, parent, opens, closes):
    """The contracts of a product alive in a window, as {instrument id: symbol}.

    Free to ask, and asked once per month bought. Their parent symbol takes in
    the calendar spreads as well - `NQM4-NQU4` comes back beside `NQM4` - and a
    spread is not a contract: it is the difference of two, with prices of its
    own and nothing to store them against. They go by the hyphen in the name.
    Leaving them out is also a third off the bill for gold, where the spreads
    are a real market, and a tenth for the indices.

    Asked for instrument ids and bought by them, because the name is not
    unique over ten years and the id is what the candles carry.
    """
    # Their range is in whole days with the end excluded, and it must not reach
    # past where the dataset goes: their history ends a few hours back, so
    # asking about the day in progress is refused outright and the day cannot
    # be asked about at all. So the range ends where the window ends, counted
    # in days, and the question is asked backwards instead of forwards: the
    # contracts alive in the days before are the contracts alive in the day
    # being bought, and one listed this very day is caught by the next run.
    #
    # A week of margin, not a day: a run on Monday morning would otherwise ask
    # about Sunday alone, and one after a holiday weekend about nothing at all.
    # Asking is free, and a contract that expired during that week comes back
    # in the list with no minutes to its name, which stores nothing.
    ends = closes.date()
    starts = min(opens.date(), ends - timedelta(days=7))
    with dbento_ask(cfg, "symbology.resolve", {
        "dataset": dataset, "symbols": parent, "stype_in": "parent",
        "stype_out": "instrument_id",
        "start_date": starts.isoformat(), "end_date": ends.isoformat(),
    }) as res:
        answer = json.load(res)
    found = {}
    for symbol, spans in (answer.get("result") or {}).items():
        if "-" in symbol:
            continue
        for span in spans or ():
            ident = (span or {}).get("s")
            if ident:
                found[str(ident)] = symbol
    return found


def contract_pair(db, cfg, feed, dataset, parent, symbol, when):
    """The row one contract hangs from, made when its first candle arrives.

    `parent` is the product it belongs to: what the roll calendar groups by,
    what says which pairs the page should leave out of its menu, and how the
    next run knows where this product left off. The symbol as they write it
    goes in the description, since the name here carries a century they do not.
    """
    return ensure(
        db, "SELECT id FROM pairs WHERE feed_id = %s AND name = %s",
        "INSERT INTO pairs (feed_id, name, description, parent) VALUES (%s, %s, %s, %s)",
        (feed, contract_name(symbol, when)),
        (f"{need(cfg, 'databento.broker')} {dataset} {symbol}", parent),
    )


def dbento_candles(rows):
    """Their CSV as rows ready to store.

    Prices are integers scaled by a thousand million and times are
    nanoseconds since the epoch, in every dataset they sell. The timestamp
    is divided as an integer: a nanosecond epoch is past what a float holds
    exactly, and dividing it as one would move the minute by a few hundred
    nanoseconds for nothing.
    """
    out = []
    for row in rows:
        try:
            seconds = int(row["ts_event"]) // DBENTO_NANOS
            price = lambda name: (Decimal(row[name]) * DBENTO_SCALE).quantize(Decimal("0.00001"))
            out.append((
                datetime.fromtimestamp(seconds, UTC),
                price("open"), price("high"), price("low"), price("close"),
                Decimal(row["volume"]),
            ))
        except (KeyError, TypeError, ValueError, ArithmeticError) as e:
            raise Fatal(f"databento: a row came out unreadable ({e}): {row}")
    return out


def dbento_feed(db, cfg):
    """A feed of its own, named after the market rather than after the shop:
    the prices are the exchange's, and buying the same instrument from
    another vendor should fill this feed rather than start a second one. Who
    sold them is the source, kept beside it."""
    broker = ensure(
        db, "SELECT id FROM brokers WHERE name = %s",
        "INSERT INTO brokers (name) VALUES (%s)",
        (need(cfg, "databento.broker"),),
    )
    market = value(cfg, "databento.market", "futures")
    feed = ensure(
        db, "SELECT id FROM feeds WHERE broker_id = %s AND environment = %s AND account_type = %s",
        "INSERT INTO feeds (broker_id, environment, account_type, market, source)"
        "  VALUES (%s, %s, %s, %s, %s)",
        (broker, "live", need(cfg, "databento.account_type")), (market, DBENTO_SOURCE),
    )
    return say_feed(db, feed, market, DBENTO_SOURCE)


def dbento_months(first, last):
    """The window cut into months, which is how it is asked for: a month is
    some thirty thousand candles, small enough to hold and long enough that
    the log moves. Cutting it costs nothing - their meter runs on the data,
    not on the number of requests."""
    at = datetime(first.year, first.month, 1, tzinfo=UTC)
    while at < last:
        nxt = datetime(at.year + at.month // 12, at.month % 12 + 1, 1, tzinfo=UTC)
        yield max(at, first), min(nxt, last)
        at = nxt


def collect_databento(db, cfg, since, until, only, spend):
    """The candles Databento sell, a month at a time.

    A symbol names either one continuous series or a whole product, and the
    two are bought differently; what they share is the window, the feed and
    the purse. What a request will cost is asked before it is made, and the
    allowance is the run's: what one symbol spends the next one no longer has.
    """
    symbols = dbento_symbols(cfg, only)
    if not symbols:
        return
    dataset = need(cfg, "databento.dataset")
    schema = value(cfg, "databento.schema", "ohlcv-1m")
    first = since or start_date(cfg, "databento.start")
    # Up to where their history reaches, which is a few hours ago and not
    # last midnight: buying only whole days left the morning on the shelf.
    last = min(until, datetime.now(UTC)) if until else datetime.now(UTC)
    ready = dbento_ready(cfg, dataset, schema)
    # Without their answer, the day before yesterday. Last midnight was the
    # rule here once and it is not safe: their history can sit hours behind
    # it - 20:21 on the previous day, one morning - and asking past where it
    # reaches is refused for the whole window, not trimmed to what exists.
    last = (min(last, ready) if ready
            else datetime.combine(last.date(), datetime.min.time(), UTC) - timedelta(days=1))
    feed = dbento_feed(db, cfg)
    db.commit()

    spent = 0.0
    for symbol in symbols:
        shape = dbento_shape(cfg, symbol)
        buy = dbento_product if shape == "parent" else dbento_series
        spent += buy(db, cfg, feed, dataset, schema, shape, symbol,
                     since, first, last, spend - spent)


def dbento_series(db, cfg, feed, dataset, schema, stype, symbol, since, first, last, allowed):
    """One series of theirs - a continuous one, or whatever a configured stype
    names - as a pair of its own. What it spent, in dollars.

    The whole window is priced in one call before any of it is bought, so a
    date typed wrong is a refusal rather than a bill.
    """
    pair = ensure(
        db, "SELECT id FROM pairs WHERE feed_id = %s AND name = %s",
        "INSERT INTO pairs (feed_id, name, description) VALUES (%s, %s, %s)",
        (feed, symbol), (f"{need(cfg, 'databento.broker')} {dataset}",),
    )
    db.commit()
    stored = db.execute(
        "SELECT max(ts) FROM bars WHERE pair_id = %s AND period = 'M1'", (pair,)
    ).fetchone()[0]
    begins = since or (stored + timedelta(minutes=1) if stored else first)
    if begins is None:
        log(f"{symbol} databento: no start date; set databento.start or run "
            f"{SELF} databento -s DATE")
        return 0.0
    if begins >= last:
        log(f"{symbol} databento: nothing new")
        return 0.0

    window = {
        "dataset": dataset, "symbols": symbol, "stype_in": stype, "schema": schema,
        "start": begins.isoformat(), "end": last.isoformat(),
    }
    costs = dbento_cost(cfg, window)
    if costs > allowed:
        raise Fatal(
            f"{symbol} databento: {begins:%Y-%m-%d} to {last:%Y-%m-%d} costs ${costs:.2f}, "
            f"over the ${allowed:.2f} this run may still spend; raise databento.spend, "
            f"or ask for less with -u DATE")
    log(f"{symbol} databento: {begins:%Y-%m-%d} to {last:%Y-%m-%d} costs ${costs:.2f}")

    received = 0
    for opens, closes in dbento_months(begins, last):
        with dbento_ask(cfg, "timeseries.get_range", dict(
            window, start=opens.isoformat(), end=closes.isoformat(), encoding="csv",
        )) as answer:
            raw = list(csv.DictReader(line.decode() for line in answer))
        rows = dbento_candles(raw)
        store(db, pair, "M1", rows)
        received += len(rows)
        if rows:
            places = dbento_digits(raw)
            say_digits(db, pair, places)
            say_pip(db, pair, places)
        if rows and received == len(rows):
            # The first candle of a run says whether the prices came out
            # at the right scale, which nothing else here can report.
            log(f"{symbol} databento: {opens:%Y-%m} opens at {rows[0][1]}")
        if TTY:
            progress(f"{symbol} databento: {opens:%Y-%m}, {received} candles")
    log(f"{symbol} databento: {received} candles, up to {last:%Y-%m-%d}")
    return costs


def dbento_product(db, cfg, feed, dataset, schema, stype, parent, since, first, last, allowed):
    """Every contract of one product, a month at a time, each contract a pair.
    What it spent, in dollars.

    Where it left off is the newest minute of any of its contracts: the front
    one trades every minute the market is open, so its frontier is the
    product's, and a contract listed later is caught by the month it starts
    trading in.

    The month is the unit for a reason beyond memory: inside one month a
    symbol of theirs means one contract and not two, which is what lets the
    century be worked out rather than asked for.

    The cost is asked per month and on the very instruments about to be
    bought, so the figure is what it will cost rather than the product's price
    with the spreads still in it. A month that would take the run past what it
    may spend is not bought at all: what came before it is kept, and the rest
    waits for the next run - how the free sources behave when their budget
    runs out, rather than an error over a bill that was never sent.
    """
    stored = db.execute(
        "SELECT max(b.ts) FROM bars b JOIN pairs p ON p.id = b.pair_id"
        " WHERE p.feed_id = %s AND p.parent = %s AND b.period = 'M1'",
        (feed, parent),
    ).fetchone()[0]
    begins = since or (stored + timedelta(minutes=1) if stored else first)
    if begins is None:
        log(f"{parent} databento: no start date; set databento.start or run "
            f"{SELF} databento -s DATE")
        return 0.0
    if begins >= last:
        log(f"{parent} databento: nothing new")
        return 0.0

    spent, received, seen, said = 0.0, 0, set(), False
    for opens, closes in dbento_months(begins, last):
        alive = dbento_contracts(cfg, dataset, parent, opens, closes)
        if not alive:
            log(f"{parent} databento: {opens:%Y-%m} names no contract of theirs")
            continue
        window = {
            "dataset": dataset, "symbols": ",".join(alive), "stype_in": "instrument_id",
            "schema": schema, "start": opens.isoformat(), "end": closes.isoformat(),
        }
        costs = dbento_cost(cfg, window)
        if spent + costs > allowed:
            log(f"{parent} databento: stopping before {opens:%Y-%m}, which costs ${costs:.2f} "
                f"of the ${allowed - spent:.2f} left to this run; raise databento.spend, "
                f"or run again to carry on from here")
            break
        with dbento_ask(cfg, "timeseries.get_range", dict(window, encoding="csv")) as answer:
            raw = list(csv.DictReader(line.decode() for line in answer))
        spent += costs

        each = {}
        for row in raw:
            each.setdefault(row.get("instrument_id"), []).append(row)
        for ident, rows in each.items():
            symbol = alive.get(ident)
            if symbol is None:
                # Bought by id, so every id in the answer is one that was
                # asked for. An unknown one means the two lists have drifted
                # apart, and guessing would file prices under another contract.
                raise Fatal(f"{parent} databento: {opens:%Y-%m} holds instrument {ident}, "
                            f"which is not among the contracts asked for")
            pair = contract_pair(db, cfg, feed, dataset, parent, symbol, opens)
            db.commit()
            bars = dbento_candles(rows)
            store(db, pair, "M1", bars)
            received += len(bars)
            seen.add(symbol)
            places = dbento_digits(rows)
            say_digits(db, pair, places)
            say_pip(db, pair, places)
            if bars and not said:
                # As for a series: the first candle of a run is where a scale
                # read wrong shows, and nothing else here would report it.
                log(f"{parent} databento: {contract_name(symbol, opens)} "
                    f"{opens:%Y-%m} opens at {bars[0][1]}")
                said = True
        if TTY:
            progress(f"{parent} databento: {opens:%Y-%m}, {len(each)} contracts, "
                     f"{received} candles, ${spent:.2f}")
        else:
            log(f"{parent} databento: {opens:%Y-%m}, {len(each)} contracts, {len(raw)} candles")
    log(f"{parent} databento: {received} candles of {len(seen)} contracts, ${spent:.2f}")
    if received:
        # The pieces of the stitched series, from the first day this run
        # touched: a roll that happened inside it is found without reading
        # sixteen years of minutes again. The cumulative factors are worked
        # out for the whole product either way, which is a hundred rows.
        pieces = db.execute(
            "SELECT rebuild_roll_steps(%s, %s)", (parent, begins.date())
        ).fetchone()[0]
        db.commit()
        # What it added, so a run that found no roll - which is every run but
        # a handful a year - says nothing rather than reporting a zero that
        # reads as a series with no pieces at all.
        if pieces:
            log(f"{parent} databento: {pieces} new pieces of the stitched series")
    return spent


def collect_only_databento(cfg, since, until, only, spend):
    """Their candles and nothing else."""
    if not dbento_symbols(cfg):
        raise Fatal("config: databento.symbols names no instrument, so there is nothing to buy")
    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        collect_databento(db, cfg, since, until, only, dbento_spend(cfg, spend))
        roll_all(db, cfg, only)


def collect_only_dukascopy(cfg, since, until, only, budget):
    """Their history and nothing else, with no budget unless one is given:
    the way to fetch twenty years in one run rather than a share per run."""
    if not dukas_pairs(cfg):
        raise Fatal("config: dukascopy.pairs names no instrument, so there is nothing to fetch")
    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        collect_dukascopy(db, cfg, since, until, only, budget or None)
        # They publish minutes and nothing else, so without this a history
        # fetched by this command alone has no hour and no day to read - and
        # no session, since the levels are built from those.
        roll_all(db, cfg, only, since)


def analyse(cfg, settings):
    """The measurements in analysis.sql, run through psql.

    Through psql rather than from here: what comes out is tables to read, and
    psql has drawn them for forty years. This adds what it does not have -
    where the file is, and which database - and gets out of the way.

    The password goes in the environment and not in the arguments, where
    anyone on the machine could read it out of the process list. The rest of
    the connection string is passed as it stands, so a setting nobody thought
    of here still reaches libpq.
    """
    if shutil.which("psql") is None:
        raise Fatal("psql not found: it comes with postgresql-client, and this command needs it")
    sql = shared("analysis.sql")
    if not os.path.isfile(sql):
        raise Fatal(f"{sql} not found")
    try:
        conn = conninfo_to_dict(need(cfg, "postgres.conninfo"))
    except psycopg.ProgrammingError as e:
        raise Fatal(f"config: postgres.conninfo: {e}")
    password = conn.pop("password", None)
    env = dict(os.environ)
    if password is not None:
        env["PGPASSWORD"] = password
    args = ["psql", make_conninfo(**conn)]
    for name, value in settings:
        args += ["-v", f"{name}={value}"]
    args += ["-f", sql]
    return subprocess.run(args, env=env, check=False).returncode


def search_pairs(cfg, token_file, text):
    configured = set(listed(cfg, "ctrader.pairs", []))
    with account_session(cfg, token_file) as client:
        available = symbols(client, need(cfg, "ctrader.account.id"))
    text = text.lower()
    found = sorted(n for n, (_, d) in available.items() if text in n.lower() or text in d.lower())
    if not found:
        raise Fatal(f"no pair of this account matches {text!r}")
    width = max(map(len, found))
    for n in found:
        print(f"{'*' if n in configured else ' '} {n:<{width}}  {available[n][1]}")


def add_pairs(path, cfg, token_file, names):
    configured = listed(cfg, "ctrader.pairs", [])
    with account_session(cfg, token_file) as client:
        available = symbols(client, need(cfg, "ctrader.account.id"))
    check_pairs(available, names, f"; {SELF} pair search looks them up")
    for n in dict.fromkeys(names):
        if n in configured:
            log(f"{n} is already configured")
    new = [n for n in dict.fromkeys(names) if n not in configured]
    if new:
        write_config(path, {"ctrader.pairs": configured + new})
        log(f"added {' '.join(new)}; {SELF} -s DATE {' '.join(f'-p {n}' for n in new)} starts the history")


HELD = """
    SELECT p.id, b.name, f.environment, f.account_type, p.name,
           s.periods, s.first_ts::date, s.last_ts::date, t.has_ticks, p.parent
    FROM pairs p
    JOIN feeds f ON f.id = p.feed_id
    JOIN brokers b ON b.id = f.broker_id
    CROSS JOIN LATERAL (
        SELECT string_agg(k.period, ' ' ORDER BY k.ordinal) AS periods,
               min(r.first_ts) AS first_ts, max(r.last_ts) AS last_ts
        FROM unnest(%s::text[]) WITH ORDINALITY AS k (period, ordinal)
        CROSS JOIN LATERAL (
            SELECT min(x.ts) AS first_ts, max(x.ts) AS last_ts
            FROM bars x WHERE x.pair_id = p.id AND x.period = k.period
        ) r
        WHERE r.first_ts IS NOT NULL
    ) s
    -- One row of the partitioned table, asked for in the primary key's own
    -- order: an EXISTS here is rewritten into a semi-join that reads it.
    CROSS JOIN LATERAL (
        SELECT count(*) > 0 AS has_ticks FROM (
            SELECT 1 FROM ticks y WHERE y.pair_id = p.id
            ORDER BY y.pair_id, y.side, y.ts LIMIT 1
        ) one
    ) t
    WHERE (%s::text[] IS NULL OR p.name = ANY(%s::text[]))
      AND (%s::text IS NULL OR lower(b.name) = lower(%s::text))
    ORDER BY b.name, f.environment, f.account_type, p.name
"""


def held(db, names=None, broker=None):
    """What each pair holds: its feed, the periods stored with the span they
    cover, and whether there are ticks behind it. All of them, or the ones
    named, or those of one broker."""
    return db.execute(HELD, (list(PERIODS), names, names, broker, broker)).fetchall()


def feed_name(broker, env, account):
    return f"{broker} {env} {account}"


# Which section of the configuration names the instruments of a broker. The
# name is the one the feed is stored under, which is where a pair comes from
# as far as anything here can tell.
def sections(cfg):
    return {
        "ctrader.pairs": value(cfg, "ctrader.account.broker"),
        "dukascopy.pairs": DUKAS_BROKER,
        "databento.symbols": value(cfg, "databento.broker"),
    }


def holding(periods, first, last, ticks):
    """What a pair holds, in a few words."""
    if periods:
        return f"{periods}{' + ticks' if ticks else ''}, {first} to {last}"
    return "ticks only" if ticks else "nothing"


def list_pairs(cfg, only_empty=False, broker=None, contracts=False):
    """The pairs the database holds, which is not the same list as the
    configuration's: one added by mistake stays until it is removed, one
    removed keeps its candles, and the page draws what is here rather than
    what is configured. So both are shown - what a pair holds, and whether a
    source still names it.

    With only_empty, the ones holding nothing at all. They are the ones the
    page has nothing to draw for and leaves out of its menu, so this is
    where they are visible.

    The contracts of a futures product are one line, the product's. A hundred
    and twenty of them would drown the list, and every one of them would read
    `no` under CONFIGURED - rightly, since what is configured is `GC.FUT` and
    not `GCZ24` - which is a hundred and twenty false alarms over the column
    that exists to show the one pair nothing names any more. With contracts
    they are listed one by one instead, with their product where that column
    would be. A product with nothing bought for it yet has no line at all,
    having no row of its own; `config show` is where it is visible.
    """
    named = set(listed(cfg, "ctrader.pairs", []) + listed(cfg, "dukascopy.pairs", [])
                + listed(cfg, "databento.symbols", []))
    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        rows = held(db, broker=broker)
    lines = []
    products = {}
    for _, broker_name, env, account, pair, periods, first, last, ticks, parent in rows:
        what = holding(periods, first, last, ticks)
        if what != "nothing" and only_empty:
            continue
        if parent and not contracts:
            products.setdefault((broker_name, env, account, parent), []).append(
                (periods, first, last))
            continue
        configured = parent if parent else "yes" if pair in named else "no"
        lines.append((broker_name, env, account, pair, configured, what))
    for (broker_name, env, account, parent), each in products.items():
        periods = sorted({p for periods, _, _ in each for p in (periods or "").split() if p},
                         key=lambda p: list(PERIODS).index(p))
        lines.append((
            broker_name, env, account, parent,
            "yes" if parent in named else "no",
            f"{len(each)} contracts, {' '.join(periods)}, "
            f"{min(first for _, first, _ in each)} to {max(last for _, _, last in each)}",
        ))
    lines.sort()
    if not lines:
        log("no pairs with nothing stored" if only_empty else "no pairs")
        return
    head = ("BROKER", "ENV", "ACCOUNT", "PAIR", "CONFIGURED", "STORED")
    width = [max(len(str(row[i])) for row in (head, *lines)) for i in range(len(head))]
    for row in (head, *lines):
        print("  ".join(str(cell).ljust(width[i]) for i, cell in enumerate(row)).rstrip())


def forget_pairs(db, rows, candles=False):
    """The database rows handed to it: gone where nothing was ever stored,
    or, with candles, gone with whatever they hold.

    A name added by mistake leaves a pair behind with no candles, which the
    page goes on offering and nothing will ever draw. One that has candles
    keeps them by default, and keeps the row they hang from: what tells the
    two apart is the database rather than an option to remember, and the
    foreign keys from bars and ticks refuse the delete in any case.

    candles is that refusal answered on purpose. It says what it is about to
    throw away and asks, once per pair, so the one irreversible thing here
    takes a terminal and a yes. The three deletes go together, ticks before
    candles before the pair.
    """
    deleted = 0
    for pair, b, e, a, name, periods, first, last, ticks, _ in rows:
        what = holding(periods, first, last, ticks)
        where = f"{feed_name(b, e, a)} {name}"
        if what == "nothing":
            db.execute("DELETE FROM pairs WHERE id = %s", (pair,))
            deleted += 1
            log(f"{where}: nothing was ever stored for it, so its pair is gone too")
        elif not candles:
            log(f"{where}: {what} stays, and so does the pair holding it")
        elif confirm(f"{where}: {what}\n  delete all of it?", False):
            db.execute("DELETE FROM ticks WHERE pair_id = %s", (pair,))
            db.execute("DELETE FROM bars WHERE pair_id = %s", (pair,))
            db.execute("DELETE FROM pairs WHERE id = %s", (pair,))
            deleted += 1
            log(f"{where}: deleted")
        else:
            log(f"{where}: kept")
    db.commit()
    return deleted


def remove_pairs(path, cfg, names, candles=False, broker=None):
    """Out of whichever source names it, and out of the database where its
    row holds nothing.

    Both halves are worth doing on their own. With three sources, a name
    taken out of one list by hand was the only way to drop a Dukascopy
    instrument or a Databento symbol; and a name the file no longer has may
    still have a row - which is what a pair added by mistake leaves behind,
    and the one somebody removing a pair can usually see. So neither half
    failing to match is an error while the other one does.

    Two sources can name the same instrument, and two feeds can hold it:
    XAUUSD is a broker's pair and a Dukascopy one. The name alone then says
    nothing about which is meant, and rather than take both - which with
    candles would throw away the wrong history - it asks for -b.

    Every question is asked before anything is written, so a refusal leaves
    the file and the database as they were rather than half done.
    """
    if candles and not sys.stdin.isatty():
        raise Fatal(f"{SELF} pair remove --candles throws data away and needs a terminal")
    keys = sections(cfg)
    if broker:
        keys = {key: who for key, who in keys.items() if who and who.lower() == broker.lower()}
    have = {key: listed(cfg, key, []) for key in keys}
    for name in dict.fromkeys(names):
        named_by = [key for key, v in have.items() if name in v]
        if len(named_by) > 1:
            raise Fatal(f"{name} is named by more than one source; -b says which\n" + "\n".join(
                f"  {keys[key]}: {key}" for key in named_by))
    removed = {key: [n for n in v if n in names] for key, v in have.items()}
    removed = {key: gone for key, gone in removed.items() if gone}

    with psycopg.connect(need(cfg, "postgres.conninfo"), connect_timeout=TIMEOUT) as db:
        rows = held(db, list(names), broker)
        for name in dict.fromkeys(names):
            same = [r for r in rows if r[4] == name]
            if len(same) > 1:
                raise Fatal(f"{name} is on more than one feed; -b says which\n" + "\n".join(
                    f"  {feed_name(b, e, a)}: {holding(periods, first, last, ticks)}"
                    for _, b, e, a, _, periods, first, last, ticks, _ in same))
        if not removed and not rows:
            whose = f" of {broker}" if broker else ""
            raise Fatal(f"nothing{whose} is called that: no source names them, and the "
                        f"database has no such pair: {' '.join(dict.fromkeys(names))}")
        if removed:
            write_config(path, {key: [n for n in have[key] if n not in names] for key in removed})
            for key, gone in removed.items():
                log(f"removed {' '.join(dict.fromkeys(gone))} from {key}")
        forget_pairs(db, rows, candles)


def crontab(user, *args):
    try:
        return subprocess.run(["crontab", "-u", user, *args], capture_output=True, text=True, check=True).stdout
    except FileNotFoundError:
        raise Fatal("crontab not found")
    except subprocess.CalledProcessError as e:
        raise Fatal(f"crontab: {e.stderr.strip()}")


def cron_user(token_file):
    """The user the collector runs as: the owner of the tokens' directory."""
    return pwd.getpwuid(dir_owner(token_file)[0]).pw_name


def crontab_lines(user):
    try:
        return crontab(user, "-l").splitlines()
    except Fatal:
        # -l fails for a user without a crontab, which is an empty one.
        return []


def install_crontab(user, lines):
    if not lines:
        try:
            crontab(user, "-r")
        except Fatal:
            pass
        return
    # Given a file rather than standard input, which not every crontab reads.
    # busybox's reads it as the user it installs for, so the file is theirs.
    entry = pwd.getpwnam(user)
    with tempfile.NamedTemporaryFile("w", suffix=".crontab") as f:
        os.fchown(f.fileno(), entry.pw_uid, entry.pw_gid)
        f.write("\n".join(lines) + "\n")
        f.flush()
        crontab(user, f.name)


def scheduled(token_file):
    """Every how many minutes the collector runs and as whom, or None."""
    user = cron_user(token_file)
    line = next((l for l in crontab_lines(user) if l.endswith(CRON_TAG)), None)
    if line is None:
        return None
    minute = line.split()[0]
    return (60 if minute == "0" else int(minute.removeprefix("*/"))), user


def need_root(command):
    if os.geteuid() != 0:
        raise Fatal(f"{SELF} {command} edits the crontab of the collector's user and needs root")


def enable(path, token_file, every):
    need_root("enable")
    user = cron_user(token_file)
    log_dir = os.path.dirname(DEFAULT_LOG)
    if not os.path.isdir(log_dir):
        # The package makes it; a run from a checkout has to make its own.
        os.makedirs(log_dir)
        os.chown(log_dir, *dir_owner(token_file))
    # An interval that does not divide the hour starts over at each hour.
    minute = "0" if every == 60 else f"*/{every}"
    command = f"{shlex.quote(os.path.realpath(sys.argv[0]))} -c {shlex.quote(os.path.abspath(path))}"
    line = f"{minute} * * * * {command} >> {DEFAULT_LOG} 2>&1 {CRON_TAG}"
    install_crontab(user, [l for l in crontab_lines(user) if not l.endswith(CRON_TAG)] + [line])
    log(f"collecting every {every} minutes as {user}, logging to {DEFAULT_LOG}")
    try:
        if subprocess.run(["pidof", "crond"], capture_output=True).returncode:
            log("crond is not running: rc-service crond start")
    except FileNotFoundError:
        pass


def disable(token_file):
    need_root("disable")
    user = cron_user(token_file)
    lines = crontab_lines(user)
    kept = [l for l in lines if not l.endswith(CRON_TAG)]
    if len(kept) == len(lines):
        log("periodic collection is not enabled")
        return
    install_crontab(user, kept)
    log(f"periodic collection disabled for {user}")


def show_config(path, cfg):
    """The configuration as the file has it, section by section and a key to
    a line, with the secrets said to be set rather than shown. The order is
    the file's own, so that reading this and editing that are the same map;
    what is not in the file - where it is, when it runs - goes at the top."""
    token_file = value(cfg, "ctrader.token_file", DEFAULT_TOKEN_FILE)
    try:
        tokens = read_tokens(token_file)
    except PermissionError:
        tokens = {"unreadable": True}
    if tokens is None:
        token_state = f"none; {SELF} ctrader token makes them"
    elif "unreadable" in tokens:
        token_state = f"{token_file}, not readable by this user"
    elif tokens.get("expires_at"):
        token_state = f"{token_file}, valid until {datetime.fromtimestamp(tokens['expires_at'], UTC):%Y-%m-%d}"
    else:
        token_state = f"{token_file}, not refreshed yet"
    if os.geteuid() != 0:
        schedule = "only root can read it"
    else:
        try:
            when = scheduled(token_file)
            schedule = f"every {when[0]} minutes as {when[1]}, log {DEFAULT_LOG}" if when else "disabled"
        except (OSError, KeyError):
            schedule = "unknown: no user owns the token directory yet"

    blocks = [(None, [("config", path), ("schedule", schedule)])]

    try:
        conn = conninfo_to_dict(value(cfg, "postgres.conninfo", ""))
    except psycopg.ProgrammingError:
        blocks.append(("postgres", [("conninfo", "not a valid connection string")]))
    else:
        named = (("host", "host"), ("port", "port"), ("database", "dbname"), ("user", "user"))
        rows = [(label, conn[key]) for label, key in named if key in conn]
        if "password" in conn:
            rows.append(("password", "set"))
        # Whatever else libpq was given - sslmode, a socket directory - after
        # the five that are asked for, since the file may hold anything.
        rows += [(k, v) for k, v in conn.items()
                 if k not in {"host", "port", "dbname", "user", "password"}]
        blocks.append(("postgres", rows or [("conninfo", "not set")]))

    names = listed(cfg, "ctrader.pairs", [])
    if not names:
        blocks.append(("cTrader", None))
    else:
        sides = listed(cfg, "ctrader.ticks.sides", [])
        since = date_text(cfg, "ctrader.ticks.start")
        blocks.append(("cTrader", [
            # Trimmed for reading: a setting that holds a space is not set
            # as far as anything here is concerned, and an empty column would
            # be read as a bug in this rather than in the file.
            ("client id", str(value(cfg, "ctrader.client_id") or "").strip() or "not set"),
            ("client secret", "set" if value(cfg, "ctrader.client_secret") else "not set"),
            ("tokens", token_state),
            ("account", value(cfg, "ctrader.account.id") or "not set"),
            ("environment", value(cfg, "ctrader.account.environment", "live")),
            ("broker", value(cfg, "ctrader.account.broker") or "not set"),
            ("account type", value(cfg, "ctrader.account.type") or "not set"),
            ("pairs", " ".join(names)),
            ("periods", " ".join(listed(cfg, "ctrader.periods", ["M1"]))),
            ("start", date_text(cfg, "ctrader.start") or "none"),
            ("ticks", "none" if not sides else ""),
            *([] if not sides else [
                ("  sides", " ".join(sides)),
                ("  start", since or "none"),
                ("  budget", f"{value(cfg, 'ctrader.ticks.budget', TICK_BUDGET)} requests a run"),
            ]),
        ]))

    theirs = listed(cfg, "dukascopy.pairs", [])
    blocks.append(("Dukascopy", None if not theirs else [
        ("pairs", " ".join(theirs)),
        ("start", date_text(cfg, "dukascopy.start") or "their first day"),
        ("budget", f"{value(cfg, 'dukascopy.budget', DUKAS_BUDGET)} days a run"),
    ]))

    bought = listed(cfg, "databento.symbols", [])
    blocks.append(("Databento", None if not bought else [
        ("key", "set" if value(cfg, "databento.key") else "NOT SET"),
        ("dataset", value(cfg, "databento.dataset") or "not set"),
        ("broker", value(cfg, "databento.broker") or "not set"),
        ("account type", value(cfg, "databento.account_type") or "not set"),
        ("market", value(cfg, "databento.market", "futures")),
        ("symbols", " ".join(bought)),
        ("schema", value(cfg, "databento.schema", "ohlcv-1m")),
        # Empty is not nothing here: it is the symbols deciding for themselves.
        ("stype", value(cfg, "databento.stype") or "by the symbol"),
        ("start", date_text(cfg, "databento.start") or "none"),
        ("spend", f"up to ${dbento_spend(cfg):.2f} a run"),
    ]))

    blocks.append((None, [("rollup", " ".join(listed(cfg, "rollup", ROLLUP)) or "none")]))

    width = max(len(k) for _, rows in blocks for k, _ in (rows or []))
    for n, (title, rows) in enumerate(blocks):
        if n:
            print()
        if title:
            print(title + ("" if rows else " (off)"))
        for key, text in rows or []:
            pad = "  " if title else ""
            print(f"{pad}{key:<{width}}  {text}".rstrip())


def ask(label, current=None, secret=False, required=False):
    """Prompt for a value; Enter keeps the current one, shown unless secret."""
    shown = None
    if current not in (None, "", 0):
        shown = "set" if secret else current
    prompt = f"  {label}" + (f" [{shown}]" if shown is not None else "") + ": "
    while True:
        answer = (getpass.getpass(prompt) if secret else input(prompt)).strip()
        v = answer or current
        if v not in (None, "", 0) or not required:
            return v
        print("  a value is needed")


def confirm(question, default):
    answer = input(f"  {question} [{'Y/n' if default else 'y/N'}]: ").strip().lower()
    return answer.startswith("y") if answer else default


def choose(label, texts, default=None):
    """Number the texts, and return the index of the one picked."""
    for i, text in enumerate(texts, 1):
        print(f"  {i:>2}) {text}")
    hint = f" [{default + 1}]" if default is not None else ""
    while True:
        answer = input(f"  {label}{hint}: ").strip()
        if not answer and default is not None:
            return default
        if answer.isdigit() and 1 <= int(answer) <= len(texts):
            return int(answer) - 1
        print(f"  a number from 1 to {len(texts)}")


def print_setup(conn):
    """The commands a superuser runs to create the database config could not
    find, with the names and the host it was given."""
    owner = conn.get("user", "ohlcdb")
    db = conn.get("dbname", "ohlcdb")
    q = shlex.quote
    server = f"-h {q(conn.get('host', 'localhost'))} -p {q(conn.get('port', '5432'))} -U SUPERUSER"
    names = f"-v db={q(db)} -v owner={q(owner)} -v reader={q(owner + '_ro')}"
    print("\nThe database is not ready. As a PostgreSQL superuser:")
    print(f"  psql {server} -d postgres {names} -f {q(shared('schema.sql'))}")
    print(f"  psql {server} -d postgres -c {q(chr(92) + 'password ' + owner)}")
    print(f"  psql {server} -d postgres -c {q(chr(92) + 'password ' + owner + '_ro')}")
    print(f"  psql {server} -d {q(db)} -f {q(shared('levels.sql'))}")


def date_text(cfg, path):
    """A date setting as the file holds it, for offering it back."""
    v = value(cfg, path)
    return str(v)[:10] if v else None


def ask_date(label, current):
    """A date the config keeps as text, or nothing when - is answered."""
    while True:
        answer = ask(f"{label}, as YYYY-MM-DD, or - for none", current)
        if answer in (None, "-"):
            return ""
        try:
            datetime.strptime(answer, "%Y-%m-%d")
            return answer
        except ValueError:
            print("  not a YYYY-MM-DD date")


def configure(path, cfg, token_file):
    if not sys.stdin.isatty():
        raise Fatal(f"{SELF} config asks questions and needs a terminal")
    new = copy.deepcopy(cfg)
    values = {}

    def setting(key, v):
        put(new, key, v)
        values[key] = v

    # The questions follow the file, section by section, so that an answer
    # can be checked against what is written where it lands. A source is a
    # block of its own behind a yes or no: answering no asks nothing of it
    # and writes none of its keys, which leaves what the file already says
    # alone rather than overwriting it with defaults.
    print("PostgreSQL")
    try:
        conn = conninfo_to_dict(value(cfg, "postgres.conninfo", ""))
    except psycopg.ProgrammingError:
        conn = {}
    db_ready = False
    while True:
        params = {
            "host": ask("host", conn.get("host"), required=True),
            "port": ask("port", conn.get("port", "5432")),
            "dbname": ask("database", conn.get("dbname", "ohlcdb")),
            "user": ask("user", conn.get("user", "ohlcdb")),
            "password": ask("password", conn.get("password"), secret=True),
        }
        # Merged over the old settings, so that those nobody asked for, such
        # as sslmode, survive.
        conn = conn | {k: v for k, v in params.items() if v is not None}
        conninfo = make_conninfo(**conn)
        try:
            with psycopg.connect(conninfo, connect_timeout=TIMEOUT) as db:
                db.execute("SELECT FROM bars LIMIT 1")
            print("  connected, the bars table is readable")
            db_ready = True
            break
        except psycopg.errors.UndefinedTable:
            # The settings are right and only the schema is behind: asking
            # for them again would not help.
            print("  connected, but the bars table is missing")
            break
        except psycopg.Error as e:
            print(f"  {str(e).strip()}")
            if not confirm("try again?", True):
                break
    setting("postgres.conninfo", conninfo)

    print("\ncTrader, a broker's own feed over the Open API")
    if confirm("configure cTrader?", True):
        print("Application, from openapi.ctrader.com")
        setting("ctrader.client_id", ask("client id", value(cfg, "ctrader.client_id"), required=True))
        setting("ctrader.client_secret",
                ask("client secret", value(cfg, "ctrader.client_secret"), secret=True, required=True))

        # The same step as the token command: the stored pair if it still
        # works, a pair asked for if it does not. Nothing of it reaches the
        # config, since a refresh spends the pair it is given. `token -n`
        # replaces a working pair; there is nothing to ask here while one
        # works - and saying no above does not touch the token file either.
        with session(new) as client:
            scope, found = accounts(client, ensure_token(client, new, token_file))
        if not found:
            raise Fatal("the token reaches no account")
        print(f"Accounts, scope {scope}")
        ids = [int(a["ctidTraderAccountId"]) for a in found]
        current = value(cfg, "ctrader.account.id")
        i = choose("account", [describe(a) for a in found], ids.index(current) if current in ids else None)
        setting("ctrader.account.id", ids[i])
        setting("ctrader.account.environment", "live" if found[i].get("isLive") else "demo")
        # A broker name edited by hand is kept as long as the account stays.
        broker = value(cfg, "ctrader.account.broker") if ids[i] == current else None
        setting("ctrader.account.broker",
                ask("broker", broker or found[i].get("brokerTitleShort"), required=True))
        setting("ctrader.account.type", ask("account type, such as Standard, Raw or ECN",
                                            value(cfg, "ctrader.account.type"), required=True))

        with account_session(new, token_file) as client:
            available = symbols(client, ids[i])
        print("Pairs, as the broker names them")
        current = " ".join(listed(cfg, "ctrader.pairs", []))
        while True:
            names = list(dict.fromkeys(ask("pairs", current, required=True).split()))
            try:
                check_pairs(available, names)
                break
            except Fatal as e:
                print(f"  {e}")
        setting("ctrader.pairs", names)

        print(f"Periods, among {' '.join(PERIODS)}")
        try:
            current = " ".join(periods(cfg))
        except Fatal:
            current = "M1"
        while True:
            chosen = ask("periods", current).upper().replace(",", " ").split()
            unknown = [p for p in chosen if p not in PERIODS]
            if chosen and not unknown:
                break
            print(f"  not a period: {' '.join(unknown)}" if unknown else "  at least one is needed")
        setting("ctrader.periods", list(dict.fromkeys(chosen)))

        print("First date to fetch for a pair or period with nothing stored yet")
        setting("ctrader.start", ask_date("start", date_text(cfg, "ctrader.start")))

        print("Ticks, the quote updates the candles are made of")
        current = " ".join(listed(cfg, "ctrader.ticks.sides", []))
        while True:
            answer = ask(f"sides, among {' '.join(SIDES)}, or - for none", current or "-")
            sides = [] if answer == "-" else answer.lower().replace(",", " ").split()
            unknown = [s for s in sides if s not in SIDES]
            if not unknown:
                break
            print(f"  not a side: {' '.join(unknown)}")
        setting("ctrader.ticks.sides", list(dict.fromkeys(sides)))
        if sides:
            setting("ctrader.ticks.start",
                    ask_date("first date to fetch", date_text(cfg, "ctrader.ticks.start")))
            while True:
                answer = ask("requests a run spends on ticks",
                             str(value(cfg, "ctrader.ticks.budget", TICK_BUDGET)))
                if str(answer).isdigit() and int(answer) >= 1:
                    break
                print("  a number of requests, one or more")
            setting("ctrader.ticks.budget", int(answer))

    print(f"\nDukascopy, free history back to 2003, stored as its own feed "
          f"({DUKAS_BROKER} {DUKAS_ACCOUNT})")
    if confirm("configure Dukascopy?", True):
        current = " ".join(listed(cfg, "dukascopy.pairs", []))
        answer = ask("their instruments, by their own names, or - for none", current or "-")
        theirs = [] if answer == "-" else list(dict.fromkeys(answer.upper().replace(",", " ").split()))
        setting("dukascopy.pairs", theirs)
        if theirs:
            # No date is an answer here: a pair with nothing stored then
            # starts where their own history does, which they are asked for.
            setting("dukascopy.start", ask_date("first date to fetch, none for as far back as they go",
                                                date_text(cfg, "dukascopy.start")))
            while True:
                answer = ask("days fetched a run", str(value(cfg, "dukascopy.budget", DUKAS_BUDGET)))
                if str(answer).isdigit() and int(answer) >= 1:
                    break
                print("  a number of days, one or more")
            setting("dukascopy.budget", int(answer))

    print("\nDatabento, who sell what the exchanges publish; the only source here "
          "that is paid for")
    if confirm("configure Databento?", True):
        setting("databento.key", ask("api key, from their portal",
                                     value(cfg, "databento.key"), secret=True, required=True))
        setting("databento.dataset", ask("dataset, such as GLBX.MDP3 for the CME",
                                         value(cfg, "databento.dataset"), required=True))
        # The feed is named after the market rather than after them, so that
        # the same instrument bought elsewhere one day fills this feed
        # instead of making a second one beside it.
        print("How the feed is named in the database: the market, not the vendor")
        setting("databento.broker", ask("broker", value(cfg, "databento.broker"), required=True))
        setting("databento.account_type", ask("account type",
                                              value(cfg, "databento.account_type"), required=True))
        while True:
            market = ask(f"market, among {' '.join(MARKETS)}",
                         value(cfg, "databento.market", "futures"), required=True).lower()
            if market in MARKETS:
                break
            print(f"  not a market: {market}")
        setting("databento.market", market)

        # schema and stype are not asked for: ohlcv-1m is the only schema
        # these tables fit, and continuous goes with the GC.v.0 shape of
        # symbol. Both are in the file for the day one of them has to change.
        current = " ".join(listed(cfg, "databento.symbols", []))
        answer = ask("their symbols, such as GC.v.0, or - for none", current or "-")
        # Left as typed: the suffix of a continuous symbol is lowercase, and
        # uppercasing it the way the Dukascopy names are would buy nothing.
        bought = [] if answer == "-" else list(dict.fromkeys(answer.replace(",", " ").split()))
        setting("databento.symbols", bought)
        if bought:
            setting("databento.start", ask_date("first date to fetch, none to wait for -s DATE",
                                                date_text(cfg, "databento.start")))
            current = value(cfg, "databento.spend", DBENTO_SPEND)
            while True:
                answer = ask("dollars a run may spend",
                             f"{current:.2f}" if isinstance(current, (int, float)) else current)
                try:
                    spend = float(answer)
                except (TypeError, ValueError):
                    spend = 0
                if spend > 0:
                    break
                print("  an amount in dollars, more than nothing")
            setting("databento.spend", spend)

    # What a source does not publish is worked out from the minutes, which is
    # why this is nobody's section. The levels read H1 and D1, so a feed of
    # minutes alone - Dukascopy's, Databento's - shows no session at all
    # until these are built.
    print("\nLengths built from M1 where the source gives none, or - for none")
    while True:
        built = ask("rollup", " ".join(listed(cfg, "rollup", ROLLUP)) or "-")
        built = [] if built == "-" else built.upper().replace(",", " ").split()
        unknown = [p for p in built if p not in PERIODS or p == "M1"]
        if not unknown:
            break
        print(f"  not a length to build: {' '.join(unknown)}")
    setting("rollup", list(dict.fromkeys(built)))

    # Everything the file will hold, not only what was answered this time: a
    # section that was skipped still collects, and is the likelier thing to
    # have forgotten.
    print(f"\nPostgreSQL {conn.get('dbname')} on {conn.get('host')}")
    if value(new, "ctrader.account.id"):
        print(f"cTrader {value(new, 'ctrader.account.id')} "
              f"{value(new, 'ctrader.account.environment')}, "
              f"{value(new, 'ctrader.account.broker')} {value(new, 'ctrader.account.type')}; "
              f"pairs {' '.join(listed(new, 'ctrader.pairs', []))}; "
              f"periods {' '.join(listed(new, 'ctrader.periods', []))}; "
              f"start {value(new, 'ctrader.start') or 'none'}")
    if theirs := listed(new, "dukascopy.pairs", []):
        print(f"Dukascopy {' '.join(theirs)}; "
              f"start {value(new, 'dukascopy.start') or 'as far back as they go'}")
    if bought := listed(new, "databento.symbols", []):
        print(f"Databento {' '.join(bought)} from {value(new, 'databento.dataset')} "
              f"as {value(new, 'databento.broker')} {value(new, 'databento.account_type')}; "
              f"start {value(new, 'databento.start') or 'none'}")
    print(f"rollup {' '.join(listed(new, 'rollup', [])) or 'none'}")
    if not confirm(f"save to {path}?", True):
        # A refresh cannot be taken back: whatever it left in the token file
        # stands, whether or not the rest of the answers are kept.
        log(f"nothing saved; the tokens in {token_file} stay as this run left them")
        return
    write_config(path, values)
    print("  saved")
    if os.geteuid() != 0:
        print(f"  {SELF} enable, run as root, collects periodically")
    elif when := scheduled(token_file):
        print(f"  collecting every {when[0]} minutes as {when[1]}")
    elif confirm(f"collect every {DEFAULT_EVERY} minutes from now on?", True):
        enable(path, token_file, DEFAULT_EVERY)
    if not db_ready:
        print_setup(conn)


def minutes_arg(v):
    n = int(v) if v.isdigit() else 0
    if not 1 <= n <= 60:
        raise argparse.ArgumentTypeError(f"{v!r} is not a number of minutes from 1 to 60")
    return n


def date_arg(v):
    try:
        return datetime.strptime(v, "%Y-%m-%d").replace(tzinfo=UTC)
    except ValueError:
        raise argparse.ArgumentTypeError(f"{v!r} is not a YYYY-MM-DD date")


def parse_args():
    # Every parser gets these, so that -c works before the command and after
    # it. SUPPRESS keeps a command's default from overwriting a -c given
    # before the command.
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("-c", "--config", metavar="FILE", default=argparse.SUPPRESS,
                        help=f'configuration file (default "{DEFAULT_CONFIG}")')

    # Apart from -c, so that it can come last in every help.
    helper = argparse.ArgumentParser(add_help=False)
    helper.add_argument("-h", "--help", action="help", help="help for %(prog)s")

    # collect is what runs without a command, so its options work there too.
    # The pairs are an option rather than positional arguments, which the
    # parser would take for a command's name.
    # One option to a parser, so that they can be handed out separately:
    # building takes no end, since a length is worked out over every minute
    # from where it starts onwards, and roll is better without -u than
    # offering one it would have to refuse.
    since = argparse.ArgumentParser(add_help=False)
    since.add_argument("-s", "--since", metavar="DATE", type=date_arg, default=argparse.SUPPRESS,
                       help="start from DATE (YYYY-MM-DD, UTC), even over candles already stored")
    until = argparse.ArgumentParser(add_help=False)
    until.add_argument("-u", "--until", metavar="DATE", type=date_arg, default=argparse.SUPPRESS,
                       help="stop at DATE (YYYY-MM-DD, UTC), which is left out")
    pairing = argparse.ArgumentParser(add_help=False)
    pairing.add_argument("-p", "--pair", metavar="PAIR", dest="pairs", action="append",
                         default=argparse.SUPPRESS,
                         help="only this configured pair; can be given more than once")
    # Apart from collecting, which the sources that hold only minutes share:
    # there is nothing to choose among in Dukascopy's history or in the ticks.
    timing = argparse.ArgumentParser(add_help=False)
    timing.add_argument("-t", "--period", metavar="PERIOD", dest="periods", action="append",
                        type=str.upper, default=argparse.SUPPRESS,
                        help="only this candle length, fetched or built; can be given more than once")
    collecting = argparse.ArgumentParser(add_help=False, parents=[since, until, pairing])
    rolling = argparse.ArgumentParser(add_help=False, parents=[since, pairing, timing])

    def command(parent, name, help, parents=(common, helper), description=None):
        return parent.add_parser(name, parents=parents, add_help=False, help=help,
                                 description=description or help[0].upper() + help[1:] + ".")

    p = argparse.ArgumentParser(description=__doc__, add_help=False,
                                parents=[common, collecting, timing, helper])
    # In brackets where the command or the action can be left out. The order
    # is the one they are used in rather than the alphabet: what is
    # configured, what is collected, what is made of it, and when it runs.
    commands = p.add_subparsers(dest="command", title="commands", metavar="[COMMAND]")

    config = command(commands, "config", "set the configuration interactively, or show it")
    config_actions = config.add_subparsers(dest="action", title="actions", metavar="[ACTION]")
    command(config_actions, "show", "print the configuration with the secrets hidden")

    pair = command(commands, "pair", "the pairs stored; search, add and remove under it")
    # Not required: the bare noun lists, and the form that cannot be typed by
    # accident is the one that changes nothing.
    pair_actions = pair.add_subparsers(dest="action", title="actions", metavar="[ACTION]")
    listing = command(pair_actions, "list",
                      "the pairs the database holds, and what each of them holds", (common,))
    listing.add_argument("-b", "--broker", metavar="NAME", help="only the pairs of this broker")
    listing.add_argument("-e", "--empty", action="store_true",
                         help="only the pairs with nothing stored")
    # Long only: -c is the configuration file, everywhere.
    listing.add_argument("--contracts", action="store_true",
                         help="one line per futures contract, not one per product")
    listing.add_argument("-h", "--help", action="help", help="help for %(prog)s")
    command(pair_actions, "search", "list the account's pairs whose name or description holds TEXT") \
        .add_argument("text", metavar="TEXT", help="what to look for, in any case")
    command(pair_actions, "add", "add pairs to the configuration, checked against the account") \
        .add_argument("names", metavar="PAIR", nargs="+", help="pair name, as the broker writes it")
    dropping = command(pair_actions, "remove",
                       "remove pairs from the configuration, and from the database where nothing was stored",
                       (common,),
                       "Take the pairs out of whichever source names them, and out of the "
                       "database where nothing was ever stored for them. What was collected "
                       "stays unless --candles says otherwise.")
    dropping.add_argument("names", metavar="PAIR", nargs="+",
                          help="pair name, as the configuration has it")
    dropping.add_argument("-b", "--broker", metavar="NAME",
                          help="the feed they belong to, where the name is on more than one")
    dropping.add_argument("--candles", action="store_true",
                          help="throw away what was collected too, asking first")
    dropping.add_argument("-h", "--help", action="help", help="help for %(prog)s")

    command(commands, "collect", "download from every source configured (the default)",
            (common, collecting, timing, helper),
            "Download what each configured source has to add since the last run.")

    # A source is a noun: on its own it collects, and what belongs to it
    # alone hangs underneath. -h by hand throughout, to keep it last.
    ctrader = command(commands, "ctrader",
                      "download the broker's candles; ticks, accounts and token under it",
                      (common, collecting, timing),
                      "Download the candles of the cTrader account, and then the share of the "
                      "ticks the configured budget allows, which is what a full collection does "
                      "for this source.")
    ctrader.add_argument("-h", "--help", action="help", help="help for %(prog)s")
    ctrader_actions = ctrader.add_subparsers(dest="action", title="actions", metavar="[ACTION]")
    ticking = command(ctrader_actions, "ticks",
                      "download only the ticks, with no budget unless one is given",
                      (common, collecting),
                      "Download the quote updates the candles are made of, holding the lock "
                      "throughout: the way to catch up on months of them in one run rather than "
                      "a share per run.")
    ticking.add_argument("-b", "--budget", metavar="REQUESTS", type=int, default=0,
                         help="stop after this many requests (default: as many as it takes)")
    ticking.add_argument("-h", "--help", action="help", help="help for %(prog)s")
    command(ctrader_actions, "accounts", "list the accounts the access token reaches")
    token = command(ctrader_actions, "token",
                    "refresh the access token, asking for a pair when needed", (common,))
    token.add_argument("-n", "--new", action="store_true",
                       help="ask for a pair from the Playground even if the stored one still works")
    token.add_argument("-h", "--help", action="help", help="help for %(prog)s")

    dukas = command(commands, "dukascopy", "download only the history Dukascopy publishes",
                    (common, collecting))
    dukas.add_argument("-b", "--budget", metavar="REQUESTS", type=int, default=0,
                       help="stop after this many days fetched (default: as many as it takes)")
    dukas.add_argument("-h", "--help", action="help", help="help for %(prog)s")

    buying = command(commands, "databento", "download only what Databento sell",
                     (common, collecting),
                     "Download the candles bought from Databento, a month at a time. What a run "
                     "would cost is asked before anything is fetched, and one that would spend "
                     "more than it is allowed stops before it starts.")
    buying.add_argument("-b", "--spend", metavar="DOLLARS", type=float, default=0,
                        help="what this run may spend (default: databento.spend)")
    buying.add_argument("-h", "--help", action="help", help="help for %(prog)s")

    command(commands, "roll", "build the longer candles from the minutes", (common, rolling, helper),
            "Build the candle lengths the sources do not publish from the M1 ones stored. "
            "A collection does this itself; run it alone to build a history collected "
            "before there was anything to build it with. Without -s, each length carries on "
            "from the last one built.")

    measuring = command(commands, "analyse", "measure the sessions with analysis.sql",
                        (common, helper),
                        "Run the measurements in analysis.sql: where the highs and lows of a "
                        "session fall among its minutes, and what becomes of the levels they "
                        "leave. It reads the candles and writes nothing. Without -b it lists "
                        "the feeds there are and stops.")
    measuring.add_argument("-b", "--broker", metavar="NAME", help="the feed to measure")
    measuring.add_argument("-p", "--pair", metavar="PAIR",
                           help="one pair of it (default: all of them together)")
    measuring.add_argument("-s", "--since", metavar="DATE", type=date_arg,
                           help="first day to measure (default: where the feed begins)")
    measuring.add_argument("-u", "--until", metavar="DATE", type=date_arg,
                           help="stop before this day (default: where the feed ends)")
    # Everything else the file declares, by its own name: a flag apiece here
    # would be a second list to keep in step with the one in the SQL.
    measuring.add_argument("-v", "--set", metavar="NAME=VALUE", dest="settings",
                           action="append", default=[],
                           help="another setting for analysis.sql: within (default 5 days), "
                                "after (60 minutes), margin (0.10 of the session range)")

    # -h is added by hand, after -e, so that it stays last.
    enabling = command(commands, "enable", "collect periodically, from the crontab of the collector's user",
                       (common,))
    enabling.add_argument("-e", "--every", metavar="MINUTES", type=minutes_arg, default=DEFAULT_EVERY,
                          help=f"minutes between runs, 1 to 60 (default {DEFAULT_EVERY})")
    enabling.add_argument("-h", "--help", action="help", help="help for %(prog)s")
    command(commands, "disable", "stop collecting periodically")

    args = p.parse_args()
    options = (("-s/--since", "since"), ("-u/--until", "until"), ("-p/--pair", "pairs"))
    given = [flag for flag, attr in options if hasattr(args, attr)]
    if given and args.command not in (None, "collect", "ctrader", "dukascopy", "databento",
                                      "roll", "analyse"):
        verb = "apply" if len(given) > 1 else "applies"
        p.error(f"{' and '.join(given)} {verb} to what collects, not to {args.command}")
    # Given before the command, or to ctrader before an action, -t is still
    # in the namespace of one that has no lengths to choose among.
    action = getattr(args, "action", None)
    if hasattr(args, "periods") and (args.command not in (None, "collect", "ctrader", "roll")
                                     or action is not None):
        what = f"{args.command} {action}" if action else args.command
        p.error(f"-t/--period applies to the candles collected and built, not to {what}")
    # getattr rather than hasattr: analyse declares both and leaves them
    # None, where the rest leave the attribute out altogether.
    if getattr(args, "since", None) and getattr(args, "until", None) and args.until <= args.since:
        p.error("-u/--until must come after -s/--since")
    if args.command == "analyse":
        named = {}
        if args.broker is not None:
            named["broker"] = args.broker
        if args.pair is not None:
            named["pair"] = args.pair
        if args.since is not None:
            named["from_day"] = args.since.date().isoformat()
        if args.until is not None:
            named["to_day"] = args.until.date().isoformat()
        rest = []
        for setting in args.settings:
            name, sep, value = setting.partition("=")
            if not sep or not name:
                p.error(f"-v takes NAME=VALUE, and {setting!r} is not one")
            if name in named:
                # psql would take the last of the two without a word, and
                # which feed a measurement is of is not a thing to guess at.
                p.error(f"-v {name} and the option that sets it say the same thing; give one")
            rest.append((name, value))
        args.settings = list(named.items()) + rest
    return args


def main():
    args = parse_args()
    path = getattr(args, "config", DEFAULT_CONFIG)
    action = getattr(args, "action", None)
    try:
        # config is how a new install gets set up, so it must not need a
        # file put there by hand first.
        if args.command == "config" and action is None and not os.path.exists(path):
            create_config(path)
        cfg = load_config(path)
        check_layout(cfg, path)
        token_file = value(cfg, "ctrader.token_file", DEFAULT_TOKEN_FILE)
        # These touch no token, so they need no lock.
        if args.command == "config" and action == "show":
            return show_config(path, cfg)
        if args.command == "pair" and action == "remove":
            return remove_pairs(path, cfg, args.names, args.candles, args.broker)
        # The bare noun shows the state; the verbs
        # are the ones that change something.
        if args.command == "pair" and action in (None, "list"):
            return list_pairs(cfg, getattr(args, "empty", False), getattr(args, "broker", None),
                              getattr(args, "contracts", False))
        if args.command == "enable":
            return enable(path, token_file, args.every)
        if args.command == "disable":
            return disable(token_file)

        if args.command == "config" or (args.command == "ctrader" and action == "token"):
            # Nor a directory for the tokens, which the lock lives beside.
            os.makedirs(os.path.dirname(token_file) or ".", mode=0o700, exist_ok=True)
        if not lock(token_file):
            log("another run is in progress, skipping")
            return
        if args.command == "config":
            configure(path, cfg, token_file)
        elif args.command == "ctrader" and action == "token":
            make_token(cfg, token_file, args.new)
        elif args.command == "ctrader" and action == "accounts":
            with session(cfg) as client:
                list_accounts(client, authorize(client, cfg, token_file))
        elif args.command == "ctrader" and action == "ticks":
            collect_only_ticks(cfg, token_file, getattr(args, "since", None),
                               getattr(args, "until", None), getattr(args, "pairs", None),
                               args.budget or None)
        elif args.command == "ctrader":
            collect_only_ctrader(cfg, token_file, getattr(args, "since", None),
                                 getattr(args, "until", None), getattr(args, "pairs", None),
                                 getattr(args, "periods", None))
        elif args.command == "databento":
            collect_only_databento(cfg, getattr(args, "since", None), getattr(args, "until", None),
                                   getattr(args, "pairs", None), args.spend)
        elif args.command == "dukascopy":
            collect_only_dukascopy(cfg, getattr(args, "since", None), getattr(args, "until", None),
                                   getattr(args, "pairs", None), args.budget)
        elif args.command == "analyse":
            return analyse(cfg, args.settings)
        elif args.command == "roll":
            roll_only(cfg, getattr(args, "since", None), getattr(args, "pairs", None),
                      getattr(args, "periods", None))
        elif args.command == "pair" and action == "search":
            search_pairs(cfg, token_file, args.text)
        elif args.command == "pair":
            add_pairs(path, cfg, token_file, args.names)
        else:
            collect(cfg, token_file, getattr(args, "since", None), getattr(args, "until", None),
                    getattr(args, "pairs", None), getattr(args, "periods", None))
    except NoTokens as e:
        # Every command but config and token stops here rather than asking:
        # collect runs from cron, where there is no one to answer.
        die(f"{e}; {SELF} ctrader token asks for a pair from the Playground and makes a new chain")
    except (Fatal, OSError, psycopg.Error, WebSocketException) as e:
        die(e)
    except (KeyboardInterrupt, EOFError):
        # Ctrl-C anywhere, or Ctrl-D at a prompt: leave the prompt's line.
        print(file=sys.stderr)
        sys.exit(130)


if __name__ == "__main__":
    main()
