trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden unmanaged position lifecycle writes

Browse Source 
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-09-12 13:31:32 +00:00
committed by GitHub
co-authored by genichin
parent 5a13c0494b
commit be2e73a2d6
2 changed files with 130 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
trex_core/adapters/portfolio/unmanaged_position_store.py
+83 -76
View File
@@ -31,82 +31,89 @@ class PostgreSQLUnmanagedPositionStore:
actor: str = "reconcile",
) -> dict:
now = observed_at or datetime.now(tz=UTC)
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
cur.execute(
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
"WHERE instance_key = %s AND symbol = %s "
"FOR UPDATE",
(instance_key, symbol),
)
row = cur.fetchone()
previous = self._row_to_state(row) if row is not None else None
previous_quantity = previous.get("quantity") if previous is not None else None
if previous is None:
transition = "detected"
cur.execute(
"INSERT INTO unmanaged_position_state "
f"({_STATE_COLUMNS}) "
"VALUES (%s, %s, %s, 'active', %s, %s, %s, NULL, "
" NULL, NULL, '', NULL, NULL, %s, %s) "
f"RETURNING {_STATE_COLUMNS}",
(instance_key, symbol, quantity, now, now, now, now, now),
)
elif previous["lifecycle_state"] == "resolved":
transition = "detected"
cur.execute(
"UPDATE unmanaged_position_state "
"SET quantity = %s, lifecycle_state = 'active', "
" first_seen_at = %s, last_seen_at = %s, quantity_changed_at = %s, "
" resolved_at = NULL, updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(quantity, now, now, now, now, instance_key, symbol),
)
elif int(previous["quantity"]) != int(quantity):
transition = "quantity_changed"
cur.execute(
"UPDATE unmanaged_position_state "
"SET quantity = %s, last_seen_at = %s, quantity_changed_at = %s, "
" resolved_at = NULL, lifecycle_state = 'active', updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(quantity, now, now, now, instance_key, symbol),
)
else:
transition = "unchanged"
cur.execute(
"UPDATE unmanaged_position_state "
"SET last_seen_at = %s, updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(now, now, instance_key, symbol),
)
state = self._row_to_state(cur.fetchone())
if transition != "unchanged":
note = (
f"previous_quantity={previous_quantity}"
if transition == "quantity_changed" and previous_quantity is not None
else ""
)
self._insert_audit(
cur,
instance_key=instance_key,
symbol=symbol,
event_type=transition,
action=None,
quantity=quantity,
lifecycle_state=state["lifecycle_state"],
actor=actor,
evidence_ref=state.get("evidence_ref"),
note=note,
created_at=now,
)
conn.commit()
return {
"transition": transition,
"previous_quantity": previous_quantity,
"state": state,
}
for _attempt in range(2):
try:
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
cur.execute(
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
"WHERE instance_key = %s AND symbol = %s "
"FOR UPDATE",
(instance_key, symbol),
)
row = cur.fetchone()
previous = self._row_to_state(row) if row is not None else None
previous_quantity = previous.get("quantity") if previous is not None else None
if previous is None:
transition = "detected"
cur.execute(
"INSERT INTO unmanaged_position_state "
f"({_STATE_COLUMNS}) "
"VALUES (%s, %s, %s, 'active', %s, %s, %s, NULL, "
" NULL, NULL, '', NULL, NULL, %s, %s) "
f"RETURNING {_STATE_COLUMNS}",
(instance_key, symbol, quantity, now, now, now, now, now),
)
elif previous["lifecycle_state"] == "resolved":
transition = "detected"
cur.execute(
"UPDATE unmanaged_position_state "
"SET quantity = %s, lifecycle_state = 'active', "
" first_seen_at = %s, last_seen_at = %s, quantity_changed_at = %s, "
" resolved_at = NULL, updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(quantity, now, now, now, now, instance_key, symbol),
)
elif int(previous["quantity"]) != int(quantity):
transition = "quantity_changed"
cur.execute(
"UPDATE unmanaged_position_state "
"SET quantity = %s, last_seen_at = %s, quantity_changed_at = %s, "
" resolved_at = NULL, lifecycle_state = 'active', updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(quantity, now, now, now, instance_key, symbol),
)
else:
transition = "unchanged"
cur.execute(
"UPDATE unmanaged_position_state "
"SET last_seen_at = %s, updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(now, now, instance_key, symbol),
)
state = self._row_to_state(cur.fetchone())
if transition != "unchanged":
note = (
f"previous_quantity={previous_quantity}"
if transition == "quantity_changed" and previous_quantity is not None
else ""
)
self._insert_audit(
cur,
instance_key=instance_key,
symbol=symbol,
event_type=transition,
action=None,
quantity=quantity,
lifecycle_state=state["lifecycle_state"],
actor=actor,
evidence_ref=state.get("evidence_ref"),
note=note,
created_at=now,
)
conn.commit()
return {
"transition": transition,
"previous_quantity": previous_quantity,
"state": state,
}
except self._psycopg.errors.UniqueViolation:
continue
raise RuntimeError(
"failed to persist unmanaged_position_state due to repeated concurrent insert race"
)
def resolve_absent(
self,