chore: run black linter

vega_query/visualisations/plots/amm.py

@@ -49,7 +49,9 @@ def create(
     # Default timestamps for getting data if required
     network_timestamp = service.api.data.get_vega_time()
     start_timestamp = (
-        market.market_timestamps.proposed if start_timestamp is None else start_timestamp
+        market.market_timestamps.proposed
+        if start_timestamp is None
+        else start_timestamp
     )
     end_timestamp = network_timestamp if end_timestamp is None else end_timestamp
 

vega_query/visualisations/plots/bbo.py

@@ -0,0 +1,129 @@
+from typing import List
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+from matplotlib.axes import Axes
+from matplotlib.figure import Figure
+
+from vega_query.service.service import Service
+import vega_protos.protos as protos
+from vega_query.utils import timestamp_to_datetime, duration_str_to_int
+from vega_query.visualisations.overlay import *
+
+from vega_query.service.networks.constants import Network
+
+
+MM_KEY = "d5ba5a505e2e5dbeb2049dae13b0150f6915cabeda52e0ccc8a8501ac672e7d9"
+
+
+def create(
+    service: Service,
+    market_code: str,
+    start_timestamp: Optional[int] = None,
+    end_timestamp: Optional[int] = None,
+):
+
+    # Get market, asset, and network parameter information
+    market = service.utils.market.find_market([market_code], include_settled=True)
+    asset = service.utils.market.find_asset([market_code])
+    epoch_length = duration_str_to_int(
+        service.api.data.get_network_parameter(key="validators.epoch.length").value
+    )
+
+    # Default timestamps for getting data if required
+    network_timestamp = service.api.data.get_vega_time()
+    start_timestamp = (
+        market.market_timestamps.proposed if start_timestamp is None else None
+    )
+    end_timestamp = network_timestamp if end_timestamp is None else None
+
+    # Get market specific information
+    market_data_history = service.api.data.get_market_data_history_by_id(
+        market_id=market.id,
+        start_timestamp=start_timestamp,
+        end_timestamp=end_timestamp,
+    )
+
+    df = service.utils.party.historic_balances(party_id=MM_KEY, asset_id=asset.id)
+
+    # import pandas as pd
+    # df = service.utils.party.historic_positions(party_id=MM_KEY, market_id=market.id)
+    # df = df.set_index(pd.to_datetime(df.index, unit='ns'))
+    # print(df.columns)
+    # plt.step(df.index, df.values, label=asset.details.name)
+    # plt.show()
+    # return
+
+    fig, ax = plt.subplots(1, 1, figsize=(15, 8))
+    twinx = ax.twinx()
+
+    overlay_best_ask_price(
+        ax,
+        market_data_history,
+        price_decimals=market.decimal_places,
+        color="red",
+    )
+    overlay_best_bid_price(
+        ax,
+        market_data_history,
+        price_decimals=market.decimal_places,
+        color="green",
+    )
+    # overlay_last_traded_price(
+    #     ax,
+    #     market_data_history,
+    #     price_decimals=market.decimal_places,
+    #     color="b",
+    # )
+    overlay_trading_mode(twinx, market_data_history)
+
+    st = fig.suptitle(f"BBO: {market.tradable_instrument.instrument.code}", ha="left")
+    fig.tight_layout()
+    return fig
+
+
+if __name__ == "__main__":
+
+    from vega_query.scripts.parser import PARSER
+
+    args = PARSER.parse_args()
+
+    # Instantiate service for specified network
+    match args.network:
+        case "mainnet":
+            network = Network.NETWORK_MAINNET
+        case "stagnet":
+            network = Network.NETWORK_STAGNET
+        case "testnet":
+            network = Network.NETWORK_TESTNET
+        case "local":
+            network = Network.NETWORK_LOCAL
+            if args.port is None:
+                raise ValueError("A port (--port) must be specified for local network.")
+        case _:
+            network = Network.NETWORK_UNSPECIFIED
+            if args.config is None:
+                raise ValueError(
+                    "A config path (--config) must be specified for unspecified network."
+                )
+    service = Service(
+        network=network,
+        network_config=args.config,
+        port_data_node=args.port,
+    )
+
+    fig = create(
+        service,
+        market_code=args.market,
+        start_timestamp=(
+            timestamp_to_datetime(args.start_time, nano=True)
+            if args.start_time is not None
+            else None
+        ),
+        end_timestamp=(
+            timestamp_to_datetime(args.end_time, nano=True)
+            if args.end_time is not None
+            else None
+        ),
+    )
+    plt.show()

vega_query/visualisations/plots/sla.py

@@ -52,7 +52,9 @@ def create(
     # Default timestamps for getting data if required
     network_timestamp = service.api.data.get_vega_time()
     start_timestamp = (
-        market.market_timestamps.proposed if start_timestamp is None else start_timestamp
+        market.market_timestamps.proposed
+        if start_timestamp is None
+        else start_timestamp
     )
     end_timestamp = network_timestamp if end_timestamp is None else end_timestamp