# -*- coding: utf-8 -*- # pylint: disable=W0212 import json import tempfile from pathlib import Path import pytest from agentscope.message import Msg class TestStorageService: def test_storage_service_defaults_to_runtime_config(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir), initial_cash=100000.0, ) assert storage.config_name == "runtime" def test_calculate_portfolio_value_cash_only(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir), initial_cash=100000.0, ) portfolio = {"cash": 100000.0, "positions": {}, "margin_used": 0.0} prices = {} value = storage.calculate_portfolio_value(portfolio, prices) assert value == 100000.0 def test_calculate_portfolio_value_with_positions(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir), initial_cash=100000.0, ) portfolio = { "cash": 50000.0, "positions": { "AAPL": {"long": 100, "short": 0}, "GOOGL": {"long": 0, "short": 10}, }, "margin_used": 5000.0, } prices = {"AAPL": 150.0, "GOOGL": 100.0} value = storage.calculate_portfolio_value(portfolio, prices) assert value == 69000.0 def test_update_dashboard_after_cycle(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir), initial_cash=100000.0, ) portfolio = { "cash": 90000.0, "positions": {"AAPL": {"long": 50, "short": 0}}, "margin_used": 0.0, } prices = {"AAPL": 200.0} storage.update_dashboard_after_cycle( portfolio=portfolio, prices=prices, date="2024-01-15", executed_trades=[ { "ticker": "AAPL", "action": "long", "quantity": 50, "price": 200.0, }, ], ) summary = storage.load_file("summary") assert summary is not None assert summary["totalAssetValue"] == 100000.0 # 90000 + 50*200 holdings = storage.load_file("holdings") assert holdings is not None assert len(holdings) > 0 trades = storage.load_file("trades") assert trades is not None assert len(trades) == 1 assert trades[0]["ticker"] == "AAPL" assert trades[0]["qty"] == 50 assert trades[0]["price"] == 200.0 def test_build_summary_export(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir), initial_cash=100000.0, ) state = { "portfolio_state": { "cash": 50000.0, "positions": {"AAPL": {"long": 100, "short": 0}}, "margin_used": 0.0, }, "equity_history": [{"t": 1000, "v": 100000}], "all_trades": [], } prices = {"AAPL": 500.0} summary = storage._build_summary_export(state, 100000.0, prices) assert summary["totalAssetValue"] == 100000.0 assert summary["totalReturn"] == 0.0 def test_build_holdings_export(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir), initial_cash=100000.0, ) state = { "portfolio_state": { "cash": 50000.0, "positions": {"AAPL": {"long": 100, "short": 0}}, "margin_used": 0.0, }, } prices = {"AAPL": 500.0} holdings = storage._build_holdings_export(state, prices) assert len(holdings) == 2 # AAPL + CASH aapl_holding = next( (h for h in holdings if h["ticker"] == "AAPL"), None, ) assert aapl_holding is not None assert aapl_holding["quantity"] == 100 assert aapl_holding["currentPrice"] == 500.0 def test_export_dashboard_compatibility_files_writes_expected_exports(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir) / "team_dashboard", initial_cash=100000.0, ) state = { "portfolio_state": { "cash": 90000.0, "positions": {"AAPL": {"long": 50, "short": 0}}, "margin_used": 0.0, }, "equity_history": [{"t": 1000, "v": 100000}], "baseline_history": [{"t": 1000, "v": 100000}], "baseline_vw_history": [{"t": 1000, "v": 100000}], "momentum_history": [{"t": 1000, "v": 100000}], "all_trades": [ { "id": "t1", "ts": 1000, "trading_date": "2024-01-15", "side": "LONG", "ticker": "AAPL", "qty": 50, "price": 200.0, } ], } prices = {"AAPL": 200.0} storage.export_dashboard_compatibility_files( state, net_value=100000.0, prices=prices, ) assert storage.load_export_file("summary")["totalAssetValue"] == 100000.0 holdings = storage.load_export_file("holdings") assert any(item["ticker"] == "AAPL" for item in holdings) assert storage.load_export_file("stats")["totalTrades"] == 1 assert storage.load_export_file("trades")[0]["ticker"] == "AAPL" def test_build_dashboard_snapshot_prefers_persisted_runtime_state_when_memory_view_is_sparse(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: dashboard_dir = Path(tmpdir) / "team_dashboard" storage = StorageService( dashboard_dir=dashboard_dir, initial_cash=100000.0, ) storage.save_server_state( { "portfolio": { "total_value": 123456.0, "cash": 45678.0, "pnl_percent": 23.45, }, "holdings": [{"ticker": "AAPL", "quantity": 10}], "stats": {"totalTrades": 3}, "trades": [{"ticker": "AAPL"}], "leaderboard": [{"agentId": "technical_analyst"}], } ) snapshot = storage.build_dashboard_snapshot_from_state({"portfolio": {}}) assert snapshot["summary"]["totalAssetValue"] == 123456.0 assert snapshot["holdings"][0]["ticker"] == "AAPL" assert snapshot["trades"][0]["ticker"] == "AAPL" assert snapshot["leaderboard"][0]["agentId"] == "technical_analyst" def test_runtime_leaderboard_prefers_server_state_and_persists_back(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: dashboard_dir = Path(tmpdir) / "team_dashboard" storage = StorageService( dashboard_dir=dashboard_dir, initial_cash=100000.0, ) storage.save_export_file("leaderboard", [{"agentId": "export_only"}]) storage.save_server_state({"leaderboard": [{"agentId": "runtime_state"}]}) leaderboard = storage.load_runtime_leaderboard() assert leaderboard[0]["agentId"] == "runtime_state" updated = [{"agentId": "updated_runtime"}] storage.persist_runtime_leaderboard(updated) saved_state = storage.read_persisted_server_state() saved_export = storage.load_export_file("leaderboard") assert saved_state["leaderboard"][0]["agentId"] == "updated_runtime" assert saved_export[0]["agentId"] == "updated_runtime" def test_compatibility_exports_can_be_disabled_without_breaking_runtime_leaderboard(self): from backend.services.storage import StorageService with tempfile.TemporaryDirectory() as tmpdir: dashboard_dir = Path(tmpdir) / "team_dashboard" storage = StorageService( dashboard_dir=dashboard_dir, initial_cash=100000.0, enable_compat_exports=False, ) storage.generate_leaderboard() storage.export_dashboard_compatibility_files( { "portfolio_state": { "cash": 100000.0, "positions": {}, "margin_used": 0.0, }, "equity_history": [], "baseline_history": [], "baseline_vw_history": [], "momentum_history": [], "all_trades": [], }, net_value=100000.0, prices={}, ) assert not dashboard_dir.joinpath("summary.json").exists() assert storage.load_runtime_leaderboard() persisted = storage.read_persisted_server_state() assert persisted["leaderboard"] def test_compatibility_exports_default_can_be_disabled_via_env(self, monkeypatch): from backend.services.storage import StorageService monkeypatch.setenv("ENABLE_DASHBOARD_COMPAT_EXPORTS", "false") with tempfile.TemporaryDirectory() as tmpdir: storage = StorageService( dashboard_dir=Path(tmpdir) / "team_dashboard", initial_cash=100000.0, ) assert storage.enable_compat_exports is False class TestTradeExecutor: def test_execute_trade_long(self): from backend.utils.trade_executor import PortfolioTradeExecutor executor = PortfolioTradeExecutor( initial_portfolio={ "cash": 100000.0, "positions": {}, "margin_requirement": 0.25, "margin_used": 0.0, }, ) result = executor.execute_trade( ticker="AAPL", action="long", quantity=10, price=150.0, ) assert result["status"] == "success" assert executor.portfolio["positions"]["AAPL"]["long"] == 10 assert executor.portfolio["cash"] == 98500.0 # 100000 - 10*150 def test_execute_trade_short(self): from backend.utils.trade_executor import PortfolioTradeExecutor executor = PortfolioTradeExecutor( initial_portfolio={ "cash": 100000.0, "positions": { "AAPL": { "long": 50, "short": 0, "long_cost_basis": 100.0, "short_cost_basis": 0.0, }, }, "margin_requirement": 0.25, "margin_used": 0.0, }, ) result = executor.execute_trade( ticker="AAPL", action="short", quantity=30, price=150.0, ) assert result["status"] == "success" assert executor.portfolio["positions"]["AAPL"]["long"] == 20 # 50 - 30 def test_execute_trade_hold(self): from backend.utils.trade_executor import PortfolioTradeExecutor executor = PortfolioTradeExecutor() result = executor.execute_trade( ticker="AAPL", action="hold", quantity=0, price=150.0, ) assert result["status"] == "success" assert result["message"] == "No trade needed" class TestPipelineExecution: def test_execute_decisions(self): """Test that pipeline executes decisions correctly. This test verifies the TradingPipeline integrates with TradeExecutor. Full integration testing is done in end-to-end tests. """ from backend.utils.trade_executor import PortfolioTradeExecutor # Use real PortfolioTradeExecutor to test the execution logic executor = PortfolioTradeExecutor( initial_portfolio={ "cash": 100000.0, "positions": {}, "margin_requirement": 0.25, "margin_used": 0.0, }, ) # Execute a long trade result = executor.execute_trade( ticker="AAPL", action="long", quantity=10, price=150.0, ) assert result["status"] == "success" assert executor.portfolio["positions"]["AAPL"]["long"] == 10 assert executor.portfolio["cash"] == 98500.0 # 100000 - 10*150 class TestMsgContentIsString: def test_msg_content_string(self): msg = Msg(name="test", content="simple string", role="user") assert isinstance(msg.content, str) def test_msg_content_json_string(self): data = {"key": "value", "nested": {"a": 1}} msg = Msg(name="test", content=json.dumps(data), role="user") assert isinstance(msg.content, str) parsed = json.loads(msg.content) assert parsed["key"] == "value" def test_msg_content_should_not_be_dict(self): data = {"key": "value"} msg = Msg(name="test", content=json.dumps(data), role="assistant") assert not isinstance(msg.content, dict) assert isinstance(msg.content, str) if __name__ == "__main__": pytest.main([__file__, "-v"])