piker/examples/trend_regimes.py

# piker: trading gear for hackers
# Copyright (C) 2018-present Tyler Goodlet (in stewardship of pikers)
# SPDX-License-Identifier: AGPL-3.0-or-later

# This program is free software: you can redistribute it and/or
# modify it under the terms of the GNU Affero General Public License
# as published by the Free Software Foundation, either version 3
# of the License, or (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public
# License along with this program. If not, see
# <https://www.gnu.org/licenses/>.

'''
Inspect synthetic timeframe evidence and optional Jev inference.

The default prints the exact request without reading credentials or
opening a network connection. ``--mock`` tests the HTTP path locally.
``--live`` sends exactly one synthetic snapshot using the explicitly
configured ``TYPESAFE_API_KEY``. No broker connection is necessary.

'''
import argparse
from dataclasses import asdict
import json
from math import floor, sin
import os
import time

import httpx
import trio

from piker.fsp._jev import (
    DEFAULT_MODEL,
    JevClassification,
    JevClient,
    build_regime_request,
)
from piker.fsp._regime import MultiTimeframeTrend, RegimeSnapshot


def synthetic_snapshot() -> RegimeSnapshot:
    '''
    Aggregate quarter-second closes ending before the current time.

    '''
    tracker: MultiTimeframeTrend = MultiTimeframeTrend(
        'synthetic.demo',
        base_period=.25,
        periods=(.25, 1, 5),
        windows=(2, 2, 2),
    )
    count: int = 960
    start: float = floor(time.time()) - count * .25
    number: int
    for number in range(count):
        price: float = (
            100 + number * .01
            + 2 * sin(number / 12)
            + .6 * sin(number / 3)
        )
        snapshot: RegimeSnapshot = tracker.push(
            start + number * .25, price,
        )
    return snapshot


def mock_reply(request: httpx.Request) -> httpx.Response:
    '''
    Demonstrate transport and parsing with explicitly fake labels.

    '''
    payload: dict = json.loads(request.content)
    answers: dict[str, dict] = {}
    name: str
    question: dict
    for name, question in payload['questions'].items():
        options: dict = question['criteria']
        probabilities: dict[str, float] = {}
        option: str
        for option in options:
            probabilities[option] = float(option == 'unknown')
        answers[name] = {
            'type': 'choice',
            'choice': 'unknown',
            'probabilities': probabilities,
            'confidence': 1.0,
        }
    return httpx.Response(200, json={
        'model': payload['model'],
        'answers': answers,
        'usage': {
            'input_tokens': 0,
            'output_tokens': 0,
        },
    })


async def classify(
    snapshot: RegimeSnapshot,
    model: str,
    live: bool,
) -> None:
    '''
    Print one explicitly live or simulated result with provenance.

    '''
    if live:
        key: str = os.environ.get('TYPESAFE_API_KEY', '')
        if not key:
            raise SystemExit('Set TYPESAFE_API_KEY before --live')
        transport: httpx.AsyncBaseTransport|None = None
    else:
        key = 'offline-demo-key'
        transport = httpx.MockTransport(mock_reply)
    http: httpx.AsyncClient
    async with httpx.AsyncClient(transport=transport) as http:
        client: JevClient = JevClient(http, key, model=model)
        result: JevClassification = await client.classify(snapshot)
    output: dict = {
        'mode': 'live_synthetic' if live else 'mock_not_inference',
        'classification': asdict(result),
    }
    print(json.dumps(output, indent=2, allow_nan=False))


def main() -> None:
    '''
    Require an explicit flag before any credential or network access.

    '''
    parser: argparse.ArgumentParser = argparse.ArgumentParser(
        description=__doc__,
    )
    mode: argparse._MutuallyExclusiveGroup = (
        parser.add_mutually_exclusive_group()
    )
    mode.add_argument(
        '--live', action='store_true',
        help='Send one synthetic TypeSafe request (uses API credit)',
    )
    mode.add_argument(
        '--mock', action='store_true',
        help='Exercise the adapter locally using simulated answers',
    )
    parser.add_argument('--model', default=DEFAULT_MODEL)
    args: argparse.Namespace = parser.parse_args()
    snapshot: RegimeSnapshot = synthetic_snapshot()
    if args.live or args.mock:
        trio.run(classify, snapshot, args.model, args.live)
    else:
        print(json.dumps(
            build_regime_request(snapshot, model=args.model),
            indent=2,
            allow_nan=False,
        ))


if __name__ == '__main__':
    main()