Spaces:
Running
Running
File size: 8,304 Bytes
715acff a64d26e 21d8407 0f123dc 156dd84 715acff 87444a0 715acff a64d26e 21d8407 0f123dc 156dd84 87444a0 715acff 87444a0 715acff a64d26e 715acff a64d26e 21d8407 0f123dc 156dd84 87444a0 715acff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 | from __future__ import annotations
from typing import Any, Dict
from src.agents.swap.storage import SwapStateRepository
from src.agents.dca.storage import DcaStateRepository
from src.agents.lending.storage import LendingStateRepository
from src.agents.staking.storage import StakingStateRepository
from src.agents.strategy.storage import StrategyStateRepository
from src.agents.liquidity.storage import LiquidityStateRepository
class Metadata:
def __init__(self):
self.crypto_data_agent: Dict[str, Any] = {}
self._swap_repo = SwapStateRepository.instance()
self._dca_repo = DcaStateRepository.instance()
self._lending_repo = LendingStateRepository.instance()
self._staking_repo = StakingStateRepository.instance()
self._strategy_repo = StrategyStateRepository.instance()
self._liquidity_repo = LiquidityStateRepository.instance()
def get_crypto_data_agent(self):
return self.crypto_data_agent
def set_crypto_data_agent(self, crypto_data_agent):
self.crypto_data_agent = crypto_data_agent
def get_swap_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._swap_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_swap_agent(
self,
swap_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if swap_agent:
self._swap_repo.set_metadata(user_id, conversation_id, swap_agent)
else:
self._swap_repo.clear_metadata(user_id, conversation_id)
except ValueError:
# Ignore clears when identity is missing; no actionable state to update.
return
def get_dca_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._dca_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_dca_agent(
self,
dca_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if dca_agent:
self._dca_repo.set_metadata(user_id, conversation_id, dca_agent)
else:
self._dca_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_dca_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._dca_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_dca_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._dca_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_swap_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._swap_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_lending_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._lending_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_lending_agent(
self,
lending_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if lending_agent:
self._lending_repo.set_metadata(user_id, conversation_id, lending_agent)
else:
self._lending_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_lending_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._lending_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_lending_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._lending_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_staking_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._staking_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_staking_agent(
self,
staking_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if staking_agent:
self._staking_repo.set_metadata(user_id, conversation_id, staking_agent)
else:
self._staking_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_staking_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._staking_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_staking_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._staking_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_strategy_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._strategy_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_strategy_agent(
self,
strategy_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if strategy_agent:
self._strategy_repo.set_metadata(user_id, conversation_id, strategy_agent)
else:
self._strategy_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_strategy_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._strategy_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_strategy_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._strategy_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
def get_liquidity_agent(self, user_id: str | None = None, conversation_id: str | None = None):
try:
return self._liquidity_repo.get_metadata(user_id, conversation_id)
except ValueError:
return {}
def set_liquidity_agent(
self,
liquidity_agent: Dict[str, Any] | None,
user_id: str | None = None,
conversation_id: str | None = None,
):
try:
if liquidity_agent:
self._liquidity_repo.set_metadata(user_id, conversation_id, liquidity_agent)
else:
self._liquidity_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def clear_liquidity_agent(self, user_id: str | None = None, conversation_id: str | None = None) -> None:
try:
self._liquidity_repo.clear_metadata(user_id, conversation_id)
except ValueError:
return
def get_liquidity_history(
self,
user_id: str | None = None,
conversation_id: str | None = None,
limit: int | None = None,
):
try:
return self._liquidity_repo.get_history(user_id, conversation_id, limit)
except ValueError:
return []
metadata = Metadata()
|