"""API function entrypoints."""
import os
import logging
import time
from importlib.metadata import version
from pathlib import Path
from typing import cast
from urllib.parse import urljoin
from pyramid.request import Request
from pyramid.response import Response, FileResponse
from pyramid.view import view_config
from tradeexecutor.cli.log import get_ring_buffer_handler
from tradeexecutor.state.metadata import Metadata
from tradeexecutor.state.store import JSONFileStore
from tradeexecutor.state.validator import validate_state_serialisation, validate_nested_state_dict
from tradeexecutor.strategy.summary import StrategySummary
from tradeexecutor.strategy.run_state import RunState
from tradeexecutor.visual.web_chart import WebChartType, render_web_chart, WebChartSource
from tradeexecutor.webhook.error import exception_response
logger = logging.getLogger(__name__)
[docs]@view_config(route_name='home', permission='view')
def web_home(request: Request):
"""/ endpoint.
The homepage displays plain text version banner.
"""
url = request.application_url
run_state: RunState = request.registry["run_state"]
# https://angrybirds.fandom.com/wiki/The_Flock
return Response(f'Chuck the Trade Executor, running strategy {run_state.executor_id}, version {run_state.version.tag}, our URL is {url}\nFor more information see https://tradingstrategy.ai\nRemember to play Angry Birds.', content_type="text/plain")
[docs]@view_config(route_name='web_ping', renderer='json', permission='view')
def web_ping(request: Request):
"""/ping endpoint
Unauthenticated endpoint to check the serverPlain is up.
"""
return {"ping": "pong"}
[docs]@view_config(route_name='web_notify', renderer='json', permission='view')
def web_notify(request: Request):
"""Notify the strategy executor about the availability of new data."""
# TODO
return {"status": "TODO"}
[docs]@view_config(route_name='web_state', renderer='json', permission='view')
def web_state(request: Request):
"""/state endpoint.
Serve the latest full state of the bog.
:return 404:
If the state has not been yet created
"""
# Does "zero copy" WSGI file serving
store: JSONFileStore = request.registry["store"]
fname = store.path
if not os.path.exists(fname):
logger.warning("Someone is eager to access the serverPlain. IP:%s, user agent:%s", request.client_addr, request.user_agent)
return exception_response(404, detail="Status file not yet created")
assert 'wsgi.file_wrapper' in request.environ, "We need wsgi.file_wrapper or we will be too slow"
r = FileResponse(content_type="application/json", request=request, path=fname)
return r
[docs]@view_config(route_name='web_status', renderer='json', permission='view')
def web_status(request: Request):
"""/status endpoint.
Return if the trade-executor is still alive or the exception that crashed it.
See :py:class:`tradeexecutor.strategy.execution_state.ExecutionState` for the return dta.
"""
run_state: RunState = request.registry["run_state"]
r = Response(content_type="application/json")
r.text = run_state.make_exportable_copy().to_json()
return r
[docs]@view_config(route_name='web_logs', renderer='json', permission='view')
def web_logs(request: Request):
"""/logs endpoint.
Return if the trade-executor is still alive or the exception that crashed it.
See :py:class:`tradeexecutor.strategy.execution_state.ExecutionState` for the return dta.
"""
ring_buffer_handler = get_ring_buffer_handler()
assert ring_buffer_handler is not None, "In-memory logging not initialised"
logs = ring_buffer_handler.export()
return logs
[docs]@view_config(route_name='web_source', permission='view')
def web_source(request: Request):
"""/source endpoint.
Return the source code of the strategy as plain text.
"""
execution_state: RunState = request.registry["run_state"]
r = Response(content_type="text/plain")
r.text = execution_state.source_code or ""
return r
[docs]@view_config(route_name='web_visualisation', permission='view')
def web_visulisation(request: Request):
"""/visualisation endpoint.
Return strategy images.
See :py:func:`tradeexecutor.strategy.pandas_trader.report_strategy_thinking` for more information.
"""
execution_state: RunState = request.registry["run_state"]
type = request.params.get("type", "small")
theme = request.params.get("theme", "light")
logger.info("Reading visualisation image, last updated %s", execution_state.visualisation.last_refreshed_at)
if type == "small":
if theme == "light":
data = execution_state.visualisation.small_image
else:
data = execution_state.visualisation.small_image_dark
if not data:
return exception_response(501, detail=f"Image data not available. It will be generated on the first strategy cycle.")
r = Response(content_type="image/svg+xml")
r.body = data
return r
elif type =="large":
if theme == "light":
data = execution_state.visualisation.large_image
else:
data = execution_state.visualisation.large_image_dark
if not data:
return exception_response(501, detail=f"Image data not available. It will be generated on the first strategy cycle.")
r = Response(content_type="image/svg+xml")
r.body = data
return r
else:
# Use 501 Not implemented error code
return exception_response(501, detail=f"Not implemented. Unknown type {type}")
[docs]@view_config(route_name='web_file', permission='view')
def web_file(request: Request):
"""/file endpoint.
Serve some trading strategy related files.
"""
metadata = cast(Metadata, request.registry["metadata"])
type = request.params.get("type")
match type:
case "notebook":
path = metadata.backtest_notebook
# https://docs.jupyter.org/en/latest/reference/mimetype.html
content_type = "application/x-ipynb+json"
case "html":
path = metadata.backtest_html
content_type = "text/html"
case _:
return exception_response(501, detail=f"Not implemented. Unknown type {type}")
if not path or not path.exists():
return exception_response(404, detail=f"Backtest data not available for {type}")
r = FileResponse(path.as_posix(), content_type=content_type)
return r
[docs]@view_config(route_name='web_chart', permission='view', renderer="json")
def web_chart(request: Request):
"""/chart endpoint.
Return chart data.
Unlike other endpoints, this endpoint does processing, albeit light.
Under wrong circumstances
"""
type_str = request.params.get("type")
source_str = request.params.get("source")
try:
type = WebChartType(type_str)
except:
return exception_response(501, detail=f"Not implemented. Unknown chart type {type_str}")
try:
source = WebChartSource(source_str)
except:
return exception_response(501, detail=f"Not implemented. Unknown source {source_str}")
if source == WebChartSource.live_trading:
store: JSONFileStore = request.registry["store"]
#: We load from the disk to prevent any
#: modify in place issues.... slow
state = store.load()
else:
metadata = cast(Metadata, request.registry["metadata"])
state = metadata.backtested_state
if not state or state.is_empty():
return exception_response(404, detail=f"Backtest data not available")
data = render_web_chart(state, type, source)
r = Response(content_type="application/json")
r.text = data.to_json()
return r
[docs]@view_config(route_name='web_icon', permission='view')
def web_icon(request: Request):
"""Render the icon of the executor."""
path = Path(os.path.join(os.path.dirname(__file__), "default_logo.png"))
assert path.exists(), f"Does not exist {path}"
r = FileResponse(path.as_posix(), content_type="image/png")
return r