File size: 6,740 Bytes
01a29cc 5d215eb 01a29cc 5d215eb 01a29cc 2faec4b 01a29cc a1d0de3 01a29cc 2faec4b 01a29cc 6a3620b 2faec4b 01a29cc 2faec4b 01a29cc 524b580 01a29cc a1d0de3 01a29cc 6a3620b 01a29cc 2faec4b 01a29cc 2faec4b 01a29cc 2faec4b 01a29cc 2faec4b 01a29cc 524b580 01a29cc a1d0de3 01a29cc | 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 | # Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
"""
Finenv Environment Implementation (Dynamic Stock via First Action)
This environment simulates stock trading.
The stock and market are NOT fixed — they are initialized dynamically
using the first action of type "init".
Actions supported:
- init → initialize stock, market, capital
- buy → buy shares
- sell → sell shares
- hold → do nothing
"""
from uuid import uuid4
from openenv.core.env_server.interfaces import Environment
from openenv.core.env_server.types import State
try:
from ..models import FinenvAction, FinenvObservation, StockExchangeMarket
from .reward import reward_message
from .stock_price import get_stock_price
except ImportError:
from models import FinenvAction, FinenvObservation, StockExchangeMarket
from server.reward import reward_message
from server.stock_price import get_stock_price
class FinenvEnvironment(Environment):
SUPPORTS_CONCURRENT_SESSIONS: bool = True
def __init__(self):
"""
Initialize environment with default values.
Actual stock configuration will be done using 'init' action.
"""
self._state = State(episode_id=str(uuid4()), step_count=0)
# Default placeholders (will be overwritten)
self.stock = None
self.market = None
self.initial_cash = 10000.0
self.max_steps = 100
self.task = "easy"
self.cash = self.initial_cash
self.shares = 0
self.price_history = []
self.stock_price = 0.0
# Flag to check if environment is initialized
self.initialized = False
def reset(self) -> FinenvObservation:
"""
Reset environment WITHOUT changing stock.
Stock will be set using first action (type="init")
"""
self._state = State(episode_id=str(uuid4()), step_count=0)
self.cash = self.initial_cash
self.shares = 0
return FinenvObservation(
stock=self.stock,
market=self.market.value if self.market else None,
price=self.stock_price,
shares=self.shares,
cash=self.cash,
done=False,
reward=reward_message(0),
metadata={"message": "Send init action to start trading"}
)
def step(self, action: FinenvAction) -> FinenvObservation:
"""
Execute one step.
First action MUST be:
{
"type": "init",
"stock": "RELIANCE",
"market": "NSE"
}
"""
if action.type == "init":
if not action.stock or not action.market:
raise ValueError("Stock and market required for init")
self.stock = action.stock
market=action.market
if isinstance(market, str):
market = market.lower()
if market == "nse":
self.market = StockExchangeMarket.NSE
elif market == "bse":
self.market = StockExchangeMarket.BSE
else:
raise ValueError("Unsupported market. Use 'NSE' or 'BSE'.")
elif isinstance(market, StockExchangeMarket):
self.market = market
else:
raise ValueError("Invalid market type. Use string or StockExchangeMarket enum.")
self.initial_cash = action.initial_cash or 10000.0
self.max_steps = action.max_steps or 100
self.task = action.task_type.lower() if action.task_type else "easy"
self.cash = self.initial_cash
self.shares = 0
# Fetch stock data
self.price_history = get_stock_price(
self.stock,
self.market.value,
self.max_steps
)
if not self.price_history:
raise ValueError("Invalid stock or no data available")
self.stock_price = self.price_history[0]
self.initialized = True
return FinenvObservation(
stock=self.stock,
market=self.market.value,
price=self.stock_price,
shares=self.shares,
cash=self.cash,
done=False,
reward=reward_message(0),
metadata={"message": "Environment initialized"}
)
if not self.initialized:
raise ValueError("Environment not initialized. Send init action first.")
# Increment step count
self._state.step_count += 1
old_value = self.cash + self.shares * self.stock_price
if action.type == "buy":
cost = action.quantity * self.stock_price
if self.cash >= cost:
self.cash -= cost
self.shares += action.quantity
elif action.type == "sell":
if self.shares >= action.quantity:
self.cash += action.quantity * self.stock_price
self.shares -= action.quantity
# hold → no action
index = min(self._state.step_count, len(self.price_history) - 1)
self.stock_price = self.price_history[index]
new_value = self.cash + self.shares * self.stock_price
reward = (new_value - old_value) / self.initial_cash
reward = max(min(reward, 1.0), -1.0)
reward=float(reward)
done = self._state.step_count >= self.max_steps
if done:
final_value = self.cash + self.shares * self.stock_price
profit = final_value - self.initial_cash
if self.task == "easy":
score = min(max(profit / (0.01 * self.initial_cash), 0), 1)
elif self.task == "medium":
score = min(max(profit / (0.05 * self.initial_cash), 0), 1)
elif self.task == "hard":
score = min(max((profit / self.initial_cash) / 0.1, 0), 1)
reward = score
reward=float(reward)
return FinenvObservation(
stock=self.stock,
market=self.market.value,
shares=self.shares,
cash=self.cash,
price=self.stock_price,
done=done,
reward=reward_message(reward),
metadata={
"step": self._state.step_count,
"portfolio_value": new_value
},
)
@property
def state(self) -> State:
return self._state |