1use super::dict;
2use crate::llm::actions::{
3 protocol_trait::{ActionResult, Protocol, Server},
4 ActionDefinition, Parameter, ParameterDefinition,
5};
6use crate::protocol::log_template::LogTemplate;
7use crate::protocol::{EventType, SpawnContext};
8use crate::state::app_state::AppState;
9use anyhow::{bail, ensure, Context, Result};
10use serde_json::{json, Value};
11use std::sync::LazyLock;
12
13#[derive(Default)]
14pub struct FixProtocol;
15impl FixProtocol {
16 pub fn new() -> Self {
17 Self
18 }
19}
20
21pub fn parameter(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
22 Parameter {
23 name: name.into(),
24 type_hint: kind.into(),
25 description: description.into(),
26 required,
27 }
28}
29
30pub fn action(
31 name: &str,
32 description: &str,
33 parameters: Vec<Parameter>,
34 example: Value,
35) -> ActionDefinition {
36 ActionDefinition {
37 name: name.into(),
38 description: description.into(),
39 parameters,
40 example,
41 log_template: Some(LogTemplate::new().with_info(format!("-> FIX {name}"))),
42 }
43}
44
45pub const BUSINESS_REJECT: &[(&str, u32)] = &[
47 ("other", 0),
48 ("unknown_id", 1),
49 ("unknown_security", 2),
50 ("unsupported_message_type", 3),
51 ("application_not_available", 4),
52 ("conditionally_required_field_missing", 5),
53 ("not_authorized", 6),
54 ("deliver_to_firm_not_available", 7),
55];
56
57pub fn send() -> ActionDefinition {
58 action(
59 "fix_send",
60 "Send one application message. Rust writes the header (BeginString, CompIDs, MsgSeqNum, SendingTime), BodyLength and CheckSum; give only body fields, in order (repeating groups as repeated entries after their count field).",
61 vec![
62 parameter("msg_type", "string", "MsgType by name or code, e.g. ExecutionReport or 8; session messages (Logon, Heartbeat, TestRequest, ResendRequest, Reject, SequenceReset, Logout) are Rust's", true),
63 parameter("fields", "array", "Body fields in order: [{\"name\": \"OrderID\", \"value\": \"O-1\"}] or [{\"tag\": 37, \"value\": \"O-1\"}]; FIX 4.4 names are accepted", true),
64 ],
65 json!({"type": "fix_send", "msg_type": "ExecutionReport", "fields": [
66 {"name": "OrderID", "value": "O-1"}, {"name": "ClOrdID", "value": "1"}, {"name": "ExecID", "value": "E-1"},
67 {"name": "ExecType", "value": "0"}, {"name": "OrdStatus", "value": "0"}, {"name": "Symbol", "value": "AAPL"},
68 {"name": "Side", "value": "1"}, {"name": "LeavesQty", "value": "100"}, {"name": "CumQty", "value": "0"}, {"name": "AvgPx", "value": "0"}
69 ]}),
70 )
71}
72fn reject() -> ActionDefinition {
73 action(
74 "fix_reject",
75 "Refuse the application message with a BusinessMessageReject (MsgType j) naming its MsgSeqNum and MsgType",
76 vec![
77 parameter("reason", "string", "other, unknown_id, unknown_security, unsupported_message_type, application_not_available, conditionally_required_field_missing, not_authorized or deliver_to_firm_not_available", true),
78 parameter("text", "string", "Text (58), up to 256 characters", true),
79 ],
80 json!({"type": "fix_reject", "reason": "unknown_security", "text": "symbol not traded here"}),
81 )
82}
83fn ignore() -> ActionDefinition {
84 action(
85 "fix_ignore",
86 "Accept the message and send nothing back (it is still counted in the session's sequence)",
87 vec![],
88 json!({"type": "fix_ignore"}),
89 )
90}
91pub fn logout() -> ActionDefinition {
92 action(
93 "fix_logout",
94 "End the session with a Logout; the connection closes when the peer answers or after a few seconds",
95 vec![parameter("text", "string", "Text (58) explaining why, up to 256 characters", false)],
96 json!({"type": "fix_logout", "text": "end of day"}),
97 )
98}
99fn accept_logon() -> ActionDefinition {
100 action(
101 "fix_accept_logon",
102 "Admit the session; Rust answers the Logon with the agreed HeartBtInt",
103 vec![],
104 json!({"type": "fix_accept_logon"}),
105 )
106}
107fn reject_logon() -> ActionDefinition {
108 action(
109 "fix_reject_logon",
110 "Refuse the session: Rust sends a Logout with this text and closes",
111 vec![parameter(
112 "text",
113 "string",
114 "Text (58), up to 256 characters",
115 true,
116 )],
117 json!({"type": "fix_reject_logon", "text": "unknown counterparty"}),
118 )
119}
120
121pub static LOGON_EVENT: LazyLock<EventType> = LazyLock::new(|| {
122 EventType::new("fix_logon", "A counterparty logs on. BeginString, CompIDs, BodyLength, CheckSum and sequence were checked by Rust.", accept_logon().example.clone())
123 .with_parameters(vec![
124 parameter("sender_comp_id", "string", "The counterparty's SenderCompID", true),
125 parameter("target_comp_id", "string", "The TargetCompID it addressed (this acceptor's)", true),
126 parameter("heartbeat_secs", "number", "HeartBtInt it asked for", true),
127 parameter("reset_seq_num", "boolean", "Whether it asked to reset sequence numbers (141=Y)", true),
128 parameter("username", "string", "Username (553), when given", false),
129 parameter("password", "string", "Password (554), when given", false),
130 ])
131 .with_actions(vec![accept_logon(), reject_logon()])
132});
133pub static MESSAGE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
134 EventType::new(
135 "fix_message",
136 "An in-sequence application message from the counterparty",
137 send().example.clone(),
138 )
139 .with_parameters(vec![
140 parameter("msg_type", "string", "MsgType code, e.g. D", true),
141 parameter(
142 "msg_type_name",
143 "string",
144 "MsgType name, e.g. NewOrderSingle",
145 true,
146 ),
147 parameter(
148 "seq",
149 "number",
150 "The message's MsgSeqNum (34) in this session, which a BusinessMessageReject refers to",
151 true,
152 ),
153 parameter(
154 "sender_comp_id",
155 "string",
156 "The counterparty's SenderCompID",
157 true,
158 ),
159 parameter(
160 "fields",
161 "array",
162 "Body fields in wire order: [{tag, name, value}]",
163 true,
164 ),
165 ])
166 .with_actions(vec![send(), reject(), ignore(), logout()])
167});
168
169pub fn body_fields(v: &Value) -> Result<Vec<(u32, String)>> {
171 let list = v["fields"]
172 .as_array()
173 .filter(|l| l.len() <= 500)
174 .context("fields is an array of at most 500 {name|tag, value}")?;
175 list.iter()
176 .map(|f| {
177 let tag = match (
178 f.get("tag").and_then(Value::as_u64),
179 f.get("name").and_then(Value::as_str),
180 ) {
181 (Some(t), _) => u32::try_from(t)
182 .ok()
183 .filter(|t| (1..=999_999).contains(t))
184 .context("tag is 1 to 999999")?,
185 (None, Some(n)) => dict::field_tag(n).with_context(|| {
186 format!("{n:?} is not a FIX 4.4 field name; give its tag number")
187 })?,
188 _ => bail!("each field has a tag or a name"),
189 };
190 ensure!(
191 !super::session::SESSION_TAGS.contains(&tag),
192 "tag {tag} belongs to the session header or trailer"
193 );
194 let value = match &f["value"] {
195 Value::String(s) => s.clone(),
196 Value::Number(n) => n.to_string(),
197 _ => bail!("tag {tag}: value is a string or number"),
198 };
199 ensure!(
200 !value.is_empty()
201 && value.len() <= 4096
202 && !crate::utils::sanitize::has_controls(&value),
203 "tag {tag}: value is 1 to 4096 printable characters"
204 );
205 Ok((tag, value))
206 })
207 .collect()
208}
209
210pub fn app_msg_type(v: &Value) -> Result<String> {
212 let raw = v["msg_type"].as_str().context("msg_type is required")?;
213 let code = dict::message_type(raw)
214 .map(str::to_owned)
215 .or_else(|| {
216 (raw.len() <= 3 && raw.bytes().all(|b| b.is_ascii_alphanumeric()))
217 .then(|| raw.to_owned())
218 })
219 .with_context(|| format!("{raw:?} is not a FIX MsgType"))?;
220 ensure!(
221 !dict::MESSAGES
222 .iter()
223 .any(|(t, _, admin)| *t == code && *admin),
224 "{raw} is a session message, which Rust sends itself"
225 );
226 Ok(code)
227}
228
229fn text_ok(v: &Value, key: &str, required: bool) -> Result<()> {
230 match v.get(key).filter(|t| !t.is_null()) {
231 None => ensure!(!required, "{key} is required"),
232 Some(t) => ensure!(
233 t.as_str().is_some_and(|t| !t.is_empty()
234 && t.len() <= 256
235 && !crate::utils::sanitize::has_controls(&t)),
236 "{key} is 1 to 256 printable characters"
237 ),
238 }
239 Ok(())
240}
241
242pub fn check_answer(v: &Value) -> Result<()> {
243 ensure!(
244 crate::utils::json_budget::within_budget(v, 1024 * 1024, 20_000, 8),
245 "answer exceeds the FIX bounds"
246 );
247 match v["type"].as_str() {
248 Some("fix_accept_logon" | "fix_ignore") => {}
249 Some("fix_reject_logon") => text_ok(v, "text", true)?,
250 Some("fix_logout") => text_ok(v, "text", false)?,
251 Some("fix_reject") => {
252 let r = v["reason"].as_str().unwrap_or_default();
253 ensure!(
254 BUSINESS_REJECT.iter().any(|(n, _)| *n == r),
255 "unknown reason {r:?}"
256 );
257 text_ok(v, "text", true)?;
258 }
259 Some("fix_send") => {
260 app_msg_type(v)?;
261 body_fields(v)?;
262 }
263 _ => bail!("Unknown FIX server action"),
264 }
265 Ok(())
266}
267
268fn startup(
269 name: &str,
270 kind: &str,
271 description: &str,
272 example: Value,
273 default: Option<Value>,
274) -> ParameterDefinition {
275 ParameterDefinition {
276 name: name.into(),
277 type_hint: kind.into(),
278 description: description.into(),
279 required: false,
280 example,
281 default,
282 }
283}
284
285impl Protocol for FixProtocol {
286 fn protocol_name(&self) -> &'static str {
287 "FIX"
288 }
289 fn stack_name(&self) -> &'static str {
290 "ETH>IP>TCP>FIX"
291 }
292 fn description(&self) -> &'static str {
293 "FIX 4.x acceptor: session layer (logon, sequence numbers, resend and gap fill, heartbeats, logout) with application messages answered by the handler"
294 }
295 fn keywords(&self) -> Vec<&'static str> {
296 vec![
297 "fix",
298 "fix protocol",
299 "financial information exchange",
300 "acceptor",
301 "order entry",
302 "execution report",
303 ]
304 }
305 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
306 vec![send(), logout()]
307 }
308 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
309 vec![
310 accept_logon(),
311 reject_logon(),
312 send(),
313 reject(),
314 ignore(),
315 logout(),
316 ]
317 }
318 fn get_event_types(&self) -> Vec<EventType> {
319 vec![LOGON_EVENT.clone(), MESSAGE_EVENT.clone()]
320 }
321 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
322 vec![
323 startup(
324 "sender_comp_id",
325 "string",
326 "This acceptor's SenderCompID; counterparties must address it as TargetCompID",
327 json!("EXCHANGE"),
328 Some(json!(super::DEFAULT_COMP_ID)),
329 ),
330 startup(
331 "begin_string",
332 "string",
333 "FIX.4.0, FIX.4.1, FIX.4.2, FIX.4.3 or FIX.4.4",
334 json!("FIX.4.2"),
335 Some(json!(super::DEFAULT_BEGIN_STRING)),
336 ),
337 startup(
338 "logon_timeout_secs",
339 "integer",
340 "Seconds a new connection has to send its Logon",
341 json!(5),
342 Some(json!(super::LOGON_TIMEOUT.as_secs())),
343 ),
344 ]
345 }
346 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
347 use crate::protocol::metadata::*;
348 ProtocolMetadataV2::builder()
349 .state(DevelopmentState::Experimental)
350 .privilege_requirement(PrivilegeRequirement::None)
351 .implementation("Hand-written FIX tag=value codec (BodyLength, CheckSum, data fields, repeated tags) and session layer shared with the client; FIX 4.4 names from the FIX44.xml dictionary QuickFIX ships")
352 .llm_control("Which counterparties may log on and how each application message is answered")
353 .e2e_testing("tests/server/fix: QuickFIX/Go 0.9.12 (independent, validating against FIX44.xml) logs on, sends NewOrderSingles answered with an ExecutionReport and a BusinessMessageReject, exchanges heartbeats and logs out, and is refused with a wrong SenderCompID")
354 .notes("One session per connection, in memory; sequence numbers restart with each connection. No FIXT.1.1/FIX 5.0, no message persistence across restarts, no encryption (EncryptMethod 0 only). Application messages are not validated against the dictionary.")
355 .answers_on_failure()
356 .max_inbound_bytes(super::codec::MAX_BODY)
357 .build()
358 }
359 fn example_prompt(&self) -> &'static str {
360 "FIX 4.4 acceptor on port 9876 that fills every limit order at its price"
361 }
362 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
363 let llm = json!({"type":"open_server","base_stack":"fix","port":9876,"instruction":"Acknowledge every NewOrderSingle with an ExecutionReport (OrdStatus New)","startup_params":{"sender_comp_id":"EXCHANGE"}});
364 let mut static_example = llm.clone();
365 static_example["event_handlers"] = json!([
366 {"event_pattern":"fix_logon","handler":{"type":"static","actions":[{"type":"fix_accept_logon"}]}},
367 {"event_pattern":"fix_message","handler":{"type":"static","actions":[{"type":"fix_reject","reason":"application_not_available","text":"orders closed"}]}}
368 ]);
369 let mut scripted = static_example.clone();
370 scripted["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nf={x['tag']:x['value'] for x in e['fields']}\nprint(json.dumps({'actions':[{'type':'fix_send','msg_type':'ExecutionReport','fields':[{'tag':37,'value':'O-'+f.get(11,'')},{'tag':11,'value':f.get(11,'')},{'tag':17,'value':'E-'+f.get(11,'')},{'tag':150,'value':'0'},{'tag':39,'value':'0'},{'tag':55,'value':f.get(55,'')},{'tag':54,'value':f.get(54,'1')},{'tag':151,'value':f.get(38,'0')},{'tag':14,'value':'0'},{'tag':6,'value':'0'}]}]}))"});
371 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
372 }
373 fn group_name(&self) -> &'static str {
374 "Application"
375 }
376}
377
378impl Server for FixProtocol {
379 fn spawn(
380 &self,
381 ctx: SpawnContext,
382 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
383 {
384 Box::pin(super::spawn(ctx))
385 }
386 fn execute_action(&self, v: Value) -> Result<ActionResult> {
387 check_answer(&v)?;
388 Ok(ActionResult::Custom {
389 name: v["type"].as_str().unwrap_or_default().into(),
390 data: v,
391 })
392 }
393}