9 Commits

Author SHA1 Message Date
298e9c39e3 Api key configuracion and quotes endpooint 2026-06-01 13:03:06 -03:00
7cc4a96a03 fix 2026-05-19 15:38:43 -03:00
1676909cbf fixes 2026-05-19 15:31:39 -03:00
d06433e0f5 add log 2026-05-19 15:15:54 -03:00
0f3ac0dd8d sending the messages by seqnum 2026-05-15 17:03:54 -03:00
4270284362 Add endpoint for all messages 2026-05-12 13:27:13 -03:00
6e46fde5d2 fixes 2026-05-11 14:27:27 -03:00
99c7f8ccb0 fixes 2026-05-11 12:34:55 -03:00
45fad9de6c generic json for fix messages 2026-05-07 17:37:17 -03:00
21 changed files with 1017 additions and 449 deletions

View File

@ -64,6 +64,9 @@ linux-build: check-env swag # Build a linux version for prod environment. Set e=
deploy: # Deploy to remote server. Set e=environment: prod, dev, demo, open-demo; s=serverName; i=instance; e.g. make deploy e=dev s=nonprodFix i=dpl deploy: # Deploy to remote server. Set e=environment: prod, dev, demo, open-demo; s=serverName; i=instance; e.g. make deploy e=dev s=nonprodFix i=dpl
make build e=$(e) && qscp build/out/distribution/qfixdpl.gz $(s):/home/quantex/qfixtb/dpl/ make build e=$(e) && qscp build/out/distribution/qfixdpl.gz $(s):/home/quantex/qfixtb/dpl/
open-demo:
make deploy e=open-demo s=nonprodFix
fmt: download-versions # Apply the Go formatter to the code fmt: download-versions # Apply the Go formatter to the code
cd tools/check; unset GOPATH; GOBIN=$$PWD/../bin go install mvdan.cc/gofumpt@$(call get_version,gofumpt); cd tools/check; unset GOPATH; GOBIN=$$PWD/../bin go install mvdan.cc/gofumpt@$(call get_version,gofumpt);
@echo "running fmt..." @echo "running fmt..."

5
docs/EndTrade.log Normal file
View File

@ -0,0 +1,5 @@
8=FIXT.1.1|9=861|35=R|34=413|49=TRADEWEB|52=20260508-16:51:56.674|56=BYMA_CORI_TEST_12345_DLRDPL|131=LST_20260508_BYMA_CORI_NY1600105.1_1|146=1|55=ARGENT 1.500 07/09/35|48=040114HT0|22=1|460=12|167=CORP|762=REGCORIINV|541=20350709|225=20200904|470=AR|223=1.5|106=ARGENTINE REPUBLIC|54=2|38=10000|64=20260511|15=USD|6110=Sov|60=20260508-16:51:56|423=1|44=-999999|236=-999999|5023=0.000010|66=NY1600105.1|6847=1|75=20260508|464=Y|20086=1|20074=Y|20075=Y|20077=[N/A]|20078=[N/A]|20079=300|20081=300|20090=0|20072=300|20098=0|5745=1|20073=RFQ|20076=Y|20156=Y|20130=20501720000|20138=HSB|561=1|562=1|20175=20210709|20223=American|22630=0|20265=LatAm|20227=Global|23068=TRWB|453=3|448=bymacust|447=C|452=3|802=3|523=BYMA CUST|803=2|523=NY|803=25|523=AR|803=4000|448=BYMA Customer|447=C|452=1|448=DTCC|447=C|452=4|454=1|455=US040114HT09|456=4|5114=2|5113=1|20169=Ca|5113=0|20169=CCC+|10=078
8=FIXT.1.1|9=203|35=AI|34=418|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:51:56.695|56=TRADEWEB|55=[N/A]|60=20260508-16:51:56.695|117=LST_20260508_BYMA_CORI_NY1600105.1_1|131=LST_20260508_BYMA_CORI_NY1600105.1_1|297=0|10=165
8=FIXT.1.1|9=292|35=AJ|34=414|49=TRADEWEB|52=20260508-16:52:05.343|56=BYMA_CORI_TEST_12345_DLRDPL|11=LST_20260508_BYMA_CORI_NY1600105.1_1|55=[N/A]|60=20260508-16:52:05|117=[N/A]|131=LST_20260508_BYMA_CORI_NY1600105.1_1|693=LST_20260508_BYMA_CORI_NY1600105.1_1_LISTEND|694=7|20086=1|20103=N/A|20110=NO|22636=Y|10=077
8=FIXT.1.1|9=211|35=AI|34=419|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:52:05.367|56=TRADEWEB|55=[N/A]|60=20260508-16:52:05.367|131=LST_20260508_BYMA_CORI_NY1600105.1_1|297=0|693=LST_20260508_BYMA_CORI_NY1600105.1_1_LISTEND|10=014

14
docs/Execution.log Normal file
View File

@ -0,0 +1,14 @@
8=FIXT.1.1|9=862|35=R|34=393|49=TRADEWEB|52=20260508-16:44:06.978|56=BYMA_CORI_TEST_12345_DLRDPL|131=LST_20260508_BYMA_CORI_NY1600095.1_1|146=1|55=ARGENT 1.500 07/09/35|48=040114HT0|22=1|460=12|167=CORP|762=REGCORIINV|541=20350709|225=20200904|470=AR|223=1.5|106=ARGENTINE REPUBLIC|54=2|38=850000|64=20260511|15=USD|6110=Sov|60=20260508-16:44:06|423=1|44=-999999|236=-999999|5023=0.000010|66=NY1600095.1|6847=1|75=20260508|464=Y|20086=1|20074=Y|20075=Y|20077=[N/A]|20078=[N/A]|20079=300|20081=300|20090=0|20072=300|20098=0|5745=1|20073=RFQ|20076=Y|20156=Y|20130=20501720000|20138=HSB|561=1|562=1|20175=20210709|20223=American|22630=0|20265=LatAm|20227=Global|23068=TRWB|453=3|448=bymacust|447=C|452=3|802=3|523=BYMA CUST|803=2|523=NY|803=25|523=AR|803=4000|448=BYMA Customer|447=C|452=1|448=DTCC|447=C|452=4|454=1|455=US040114HT09|456=4|5114=2|5113=1|20169=Ca|5113=0|20169=CCC+|10=163
8=FIXT.1.1|9=203|35=AI|34=398|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:44:07.018|56=TRADEWEB|55=[N/A]|60=20260508-16:44:07.018|117=LST_20260508_BYMA_CORI_NY1600095.1_1|131=LST_20260508_BYMA_CORI_NY1600095.1_1|297=0|10=162
8=FIXT.1.1|9=293|35=S|34=399|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:44:34.096|56=TRADEWEB|15=USD|22=1|38=850000|44=91.00000000|48=040114HT0|54=2|55=[N/A]|60=20260508-16:44:34.096|64=20260511|117=LST_20260508_BYMA_CORI_NY1600095.1_1|131=LST_20260508_BYMA_CORI_NY1600095.1_1|132=91.00000000|423=1|537=211|10=127
8=FIXT.1.1|9=191|35=CW|34=394|49=TRADEWEB|52=20260508-16:44:34.137|56=BYMA_CORI_TEST_12345_DLRDPL|60=20260508-16:44:34|117=LST_20260508_BYMA_CORI_NY1600095.1_1|131=LST_20260508_BYMA_CORI_NY1600095.1_1|1865=1|10=004
8=FIXT.1.1|9=80|35=0|34=400|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:45:04.125|56=TRADEWEB|10=169
8=FIXT.1.1|9=80|35=0|34=395|49=TRADEWEB|52=20260508-16:45:04.142|56=BYMA_CORI_TEST_12345_DLRDPL|10=181
8=FIXT.1.1|9=391|35=AJ|34=396|49=TRADEWEB|52=20260508-16:45:10.511|56=BYMA_CORI_TEST_12345_DLRDPL|11=LST_20260508_BYMA_CORI_NY1600095.1_1|22=1|38=850000|44=91|48=040114HT0|54=2|55=[N/A]|60=20260508-16:45:10|117=LST_20260508_BYMA_CORI_NY1600095.1_1|131=LST_20260508_BYMA_CORI_NY1600095.1_1|423=1|693=LST_20260508_BYMA_CORI_NY1600095.1_1_TRDREQ|694=1|20074=N|20075=N|20076=N|20079=236|20082=60|20156=N|22630=0|10=127
8=FIXT.1.1|9=295|35=8|34=397|49=TRADEWEB|52=20260508-16:45:10.514|56=BYMA_CORI_TEST_12345_DLRDPL|6=0|11=LST_20260508_BYMA_CORI_NY1600095.1_1|14=0|17=LST_20260508_BYMA_CORI_NY1600095.1_1_LISTEND-124510.514|37=LST_20260508_BYMA_CORI_NY1600095.1_1|39=A|55=[N/A]|60=20260508-16:45:10|75=20260508|150=A|151=0|20086=1|10=213
8=FIXT.1.1|9=210|35=AI|34=401|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:45:10.539|56=TRADEWEB|55=[N/A]|60=20260508-16:45:10.539|131=LST_20260508_BYMA_CORI_NY1600095.1_1|297=0|693=LST_20260508_BYMA_CORI_NY1600095.1_1_TRDREQ|10=209
8=FIXT.1.1|9=261|35=BN|34=402|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:45:10.601|56=TRADEWEB|11=LST_20260508_BYMA_CORI_NY1600095.1_1|17=LST_20260508_BYMA_CORI_NY1600095.1_1_LISTEND-124510.514|37=LST_20260508_BYMA_CORI_NY1600095.1_1|55=[N/A]|60=20260508-16:45:10.601|1036=1|10=232
8=FIXT.1.1|9=303|35=8|34=398|49=TRADEWEB|52=20260508-16:45:10.666|56=BYMA_CORI_TEST_12345_DLRDPL|6=0|11=LST_20260508_BYMA_CORI_NY1600095.1_1|14=0|17=LST_20260508_BYMA_CORI_NY1600095.1_1_TRDEND-124510.660|37=LST_20260508_BYMA_CORI_NY1600095.1_1|39=2|44=91|54=2|55=[N/A]|60=20260508-16:45:10|75=20260508|150=F|151=0|423=1|10=252
8=FIXT.1.1|9=260|35=BN|34=403|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:45:10.690|56=TRADEWEB|11=LST_20260508_BYMA_CORI_NY1600095.1_1|17=LST_20260508_BYMA_CORI_NY1600095.1_1_TRDEND-124510.660|37=LST_20260508_BYMA_CORI_NY1600095.1_1|55=[N/A]|60=20260508-16:45:10.690|1036=1|10=168
8=FIXT.1.1|9=729|35=8|34=399|49=TRADEWEB|52=20260508-16:45:10.738|56=BYMA_CORI_TEST_12345_DLRDPL|6=0|11=LST_20260508_BYMA_CORI_NY1600095.1_1|14=0|17=LST_20260508_BYMA_CORI_NY1600095.1_1_TRDSUMM-124510.736|22=1|37=LST_20260508_BYMA_CORI_NY1600095.1_1|38=850000|39=2|44=91|48=040114HT0|54=2|55=[N/A]|60=20260508-16:45:10.535|64=20260511|75=20260508|150=F|151=0|167=CORP|236=2.61081622|423=1|453=2|448=bymacust|447=C|452=3|802=3|523=BYMA CUST|803=2|523=NY|803=25|523=AR|803=4000|448=BYMA Customer|447=C|452=1|802=1|523=YES|803=4003|454=1|455=US040114HT09|456=4|526=TRD_20260508_BYMA_CORI_213|1003=20260508.BYMA.CORI.213|1907=1|1903=20260508BYMACORI213|1906=5|6731=20260508.BYMA.CORI.213|20115=91|22630=0|22636=Y|23068=TRWB|23096=20260508BYMACORI213|10=051
8=FIXT.1.1|9=261|35=BN|34=404|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:45:10.781|56=TRADEWEB|11=LST_20260508_BYMA_CORI_NY1600095.1_1|17=LST_20260508_BYMA_CORI_NY1600095.1_1_TRDSUMM-124510.736|37=LST_20260508_BYMA_CORI_NY1600095.1_1|55=[N/A]|60=20260508-16:45:10.781|1036=1|10=027

10
docs/Timeout.log Normal file
View File

@ -0,0 +1,10 @@
8=FIXT.1.1|9=858|35=R|34=416|49=TRADEWEB|52=20260508-16:52:50.205|56=BYMA_CORI_TEST_12345_DLRDPL|131=LST_20260508_BYMA_CORI_NY1600105.2_1|146=1|55=ARGENT 1.500 07/09/35|48=040114HT0|22=1|460=12|167=CORP|762=REGCORIINV|541=20350709|225=20200904|470=AR|223=1.5|106=ARGENTINE REPUBLIC|54=2|38=15000|64=20260511|15=USD|6110=Sov|60=20260508-16:52:50|423=1|44=-999999|236=-999999|5023=0.000010|66=NY1600105.2|6847=1|75=20260508|464=Y|20086=1|20074=Y|20075=Y|20077=[N/A]|20078=[N/A]|20079=60|20081=60|20090=0|20072=60|20098=0|5745=1|20073=RFQ|20076=Y|20156=Y|20130=20501720000|20138=HSB|561=1|562=1|20175=20210709|20223=American|22630=0|20265=LatAm|20227=Global|23068=TRWB|453=3|448=bymacust|447=C|452=3|802=3|523=BYMA CUST|803=2|523=NY|803=25|523=AR|803=4000|448=BYMA Customer|447=C|452=1|448=DTCC|447=C|452=4|454=1|455=US040114HT09|456=4|5114=2|5113=1|20169=Ca|5113=0|20169=CCC+|10=195
8=FIXT.1.1|9=203|35=AI|34=421|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:52:50.226|56=TRADEWEB|55=[N/A]|60=20260508-16:52:50.226|117=LST_20260508_BYMA_CORI_NY1600105.2_1|131=LST_20260508_BYMA_CORI_NY1600105.2_1|297=0|10=131
8=FIXT.1.1|9=80|35=0|34=422|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:53:20.289|56=TRADEWEB|10=181
8=FIXT.1.1|9=80|35=0|34=417|49=TRADEWEB|52=20260508-16:53:20.291|56=BYMA_CORI_TEST_12345_DLRDPL|10=178
8=FIXT.1.1|9=80|35=0|34=423|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:53:50.295|56=TRADEWEB|10=182
8=FIXT.1.1|9=80|35=0|34=418|49=TRADEWEB|52=20260508-16:53:50.305|56=BYMA_CORI_TEST_12345_DLRDPL|10=178
8=FIXT.1.1|9=265|35=AJ|34=419|49=TRADEWEB|52=20260508-16:53:50.698|56=BYMA_CORI_TEST_12345_DLRDPL|11=LST_20260508_BYMA_CORI_NY1600105.2_1|55=[N/A]|60=20260508-16:53:50|117=[N/A]|131=LST_20260508_BYMA_CORI_NY1600105.2_1|693=LST_20260508_BYMA_CORI_NY1600105.2_1_LISTEND|694=8|20086=1|10=010
8=FIXT.1.1|9=211|35=AI|34=424|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:53:50.721|56=TRADEWEB|55=[N/A]|60=20260508-16:53:50.721|131=LST_20260508_BYMA_CORI_NY1600105.2_1|297=0|693=LST_20260508_BYMA_CORI_NY1600105.2_1_LISTEND|10=002
8=FIXT.1.1|9=292|35=AJ|34=420|49=TRADEWEB|52=20260508-16:53:53.741|56=BYMA_CORI_TEST_12345_DLRDPL|11=LST_20260508_BYMA_CORI_NY1600105.2_1|55=[N/A]|60=20260508-16:53:53|117=[N/A]|131=LST_20260508_BYMA_CORI_NY1600105.2_1|693=LST_20260508_BYMA_CORI_NY1600105.2_1_LISTEND|694=8|20086=1|20103=N/A|20110=NO|22636=Y|10=088
8=FIXT.1.1|9=211|35=AI|34=425|49=BYMA_CORI_TEST_12345_DLRDPL|52=20260508-16:53:53.763|56=TRADEWEB|55=[N/A]|60=20260508-16:53:53.763|131=LST_20260508_BYMA_CORI_NY1600105.2_1|297=0|693=LST_20260508_BYMA_CORI_NY1600105.2_1_LISTEND|10=021

View File

@ -5,6 +5,10 @@ SenderCompID=QUANTEX
ResetOnLogon=Y ResetOnLogon=Y
FileStorePath=fix_store FileStorePath=fix_store
FileLogPath=fix_logs FileLogPath=fix_logs
TransportDataDictionary=spec/FIXT11.xml
AppDataDictionary=spec/FIX50SP2.xml
AllowUnknownMessageFields=Y
RejectInvalidMessage=N
[SESSION] [SESSION]
BeginString=FIXT.1.1 BeginString=FIXT.1.1

View File

@ -89,6 +89,24 @@ func (m FieldMap) Get(parser Field) MessageRejectError {
return m.GetField(parser.Tag(), parser) return m.GetField(parser.Tag(), parser)
} }
// RawValues returns a copy of the underlying TagValue slice for a tag.
// For repeating groups, the first entry is the count and the remaining
// entries are the flattened inner-field values across all repetitions.
func (m FieldMap) RawValues(tag Tag) []TagValue {
m.rwLock.RLock()
defer m.rwLock.RUnlock()
f, ok := m.tagLookup[tag]
if !ok {
return nil
}
out := make([]TagValue, len(f))
copy(out, f)
return out
}
// Has returns true if the Tag is present in this FieldMap. // Has returns true if the Tag is present in this FieldMap.
func (m FieldMap) Has(tag Tag) bool { func (m FieldMap) Has(tag Tag) bool {
m.rwLock.RLock() m.rwLock.RLock()

View File

@ -86,6 +86,16 @@ func (tv TagValue) String() string {
return string(tv.bytes) return string(tv.bytes)
} }
// Tag returns the FIX tag for this TagValue.
func (tv TagValue) Tag() Tag {
return tv.tag
}
// Value returns the raw FIX value bytes for this TagValue.
func (tv TagValue) Value() []byte {
return tv.value
}
func bytesTotal(bytes []byte) (total int) { func bytesTotal(bytes []byte) (total int) {
for _, b := range bytes { for _, b := range bytes {
total += int(b) total += int(b)

313
spec/FIXT11.xml Normal file
View File

@ -0,0 +1,313 @@
<?xml version='1.0' encoding='UTF-8'?>
<fix type='FIXT' major='1' minor='1' servicepack='0'>
<header>
<field name='BeginString' required='Y'/>
<field name='BodyLength' required='Y'/>
<field name='MsgType' required='Y'/>
<field name='SenderCompID' required='Y'/>
<field name='TargetCompID' required='Y'/>
<field name='OnBehalfOfCompID' required='N'/>
<field name='DeliverToCompID' required='N'/>
<field name='SecureDataLen' required='N'/>
<field name='SecureData' required='N'/>
<field name='MsgSeqNum' required='Y'/>
<field name='SenderSubID' required='N'/>
<field name='SenderLocationID' required='N'/>
<field name='TargetSubID' required='N'/>
<field name='TargetLocationID' required='N'/>
<field name='OnBehalfOfSubID' required='N'/>
<field name='OnBehalfOfLocationID' required='N'/>
<field name='DeliverToSubID' required='N'/>
<field name='DeliverToLocationID' required='N'/>
<field name='PossDupFlag' required='N'/>
<field name='PossResend' required='N'/>
<field name='SendingTime' required='Y'/>
<field name='OrigSendingTime' required='N'/>
<field name='XmlDataLen' required='N'/>
<field name='XmlData' required='N'/>
<field name='MessageEncoding' required='N'/>
<field name='LastMsgSeqNumProcessed' required='N'/>
<component name='HopGrp' required='N'/>
<field name='ApplVerID' required='N'/>
<field name='CstmApplVerID' required='N'/>
</header>
<trailer>
<field name='SignatureLength' required='N'/>
<field name='Signature' required='N'/>
<field name='CheckSum' required='Y'/>
</trailer>
<messages>
<message msgcat='admin' msgtype='0' name='Heartbeat'>
<field name='TestReqID' required='N'/>
</message>
<message msgcat='admin' msgtype='1' name='TestRequest'>
<field name='TestReqID' required='Y'/>
</message>
<message msgcat='admin' msgtype='2' name='ResendRequest'>
<field name='BeginSeqNo' required='Y'/>
<field name='EndSeqNo' required='Y'/>
</message>
<message msgcat='admin' msgtype='3' name='Reject'>
<field name='RefSeqNum' required='Y'/>
<field name='RefTagID' required='N'/>
<field name='RefMsgType' required='N'/>
<field name='SessionRejectReason' required='N'/>
<field name='Text' required='N'/>
<field name='EncodedTextLen' required='N'/>
<field name='EncodedText' required='N'/>
</message>
<message msgcat='admin' msgtype='4' name='SequenceReset'>
<field name='GapFillFlag' required='N'/>
<field name='NewSeqNo' required='Y'/>
</message>
<message msgcat='admin' msgtype='5' name='Logout'>
<field name='Text' required='N'/>
<field name='EncodedTextLen' required='N'/>
<field name='EncodedText' required='N'/>
</message>
<message msgcat='admin' msgtype='A' name='Logon'>
<field name='EncryptMethod' required='Y'/>
<field name='HeartBtInt' required='Y'/>
<field name='RawDataLength' required='N'/>
<field name='RawData' required='N'/>
<field name='ResetSeqNumFlag' required='N'/>
<field name='NextExpectedMsgSeqNum' required='N'/>
<field name='MaxMessageSize' required='N'/>
<field name='TestMessageIndicator' required='N'/>
<field name='Username' required='N'/>
<field name='Password' required='N'/>
<field name='DefaultApplVerID' required='Y'/>
<component name='MsgTypeGrp' required='N'/>
</message>
</messages>
<components>
<component name='HopGrp'>
<group name='NoHops' required='N'>
<field name='HopCompID' required='N'/>
<field name='HopSendingTime' required='N'/>
<field name='HopRefID' required='N'/>
</group>
</component>
<component name='MsgTypeGrp'>
<group name='NoMsgTypes' required='N'>
<field name='RefMsgType' required='N'/>
<field name='MsgDirection' required='N'/>
<field name='RefApplVerID' required='N'/>
<field name='RefCstmApplVerID' required='N'/>
</group>
</component>
</components>
<fields>
<field name='BeginSeqNo' number='7' type='SEQNUM'/>
<field name='BeginString' number='8' type='STRING'/>
<field name='BodyLength' number='9' type='LENGTH'/>
<field name='CheckSum' number='10' type='STRING'/>
<field name='EndSeqNo' number='16' type='SEQNUM'/>
<field name='MsgSeqNum' number='34' type='SEQNUM'/>
<field number='35' name='MsgType' type='STRING'>
<value enum='0' description='HEARTBEAT'/>
<value enum='1' description='TEST_REQUEST'/>
<value enum='2' description='RESEND_REQUEST'/>
<value enum='3' description='REJECT'/>
<value enum='4' description='SEQUENCE_RESET'/>
<value enum='5' description='LOGOUT'/>
<value enum='6' description='INDICATION_OF_INTEREST'/>
<value enum='7' description='ADVERTISEMENT'/>
<value enum='8' description='EXECUTION_REPORT'/>
<value enum='9' description='ORDER_CANCEL_REJECT'/>
<value enum='A' description='LOGON'/>
<value enum='B' description='NEWS'/>
<value enum='C' description='EMAIL'/>
<value enum='D' description='ORDER_SINGLE'/>
<value enum='E' description='ORDER_LIST'/>
<value enum='F' description='ORDER_CANCEL_REQUEST'/>
<value enum='G' description='ORDER_CANCEL_REPLACE_REQUEST'/>
<value enum='H' description='ORDER_STATUS_REQUEST'/>
<value enum='J' description='ALLOCATION_INSTRUCTION'/>
<value enum='K' description='LIST_CANCEL_REQUEST'/>
<value enum='L' description='LIST_EXECUTE'/>
<value enum='M' description='LIST_STATUS_REQUEST'/>
<value enum='N' description='LIST_STATUS'/>
<value enum='P' description='ALLOCATION_INSTRUCTION_ACK'/>
<value enum='Q' description='DONT_KNOW_TRADE'/>
<value enum='R' description='QUOTE_REQUEST'/>
<value enum='S' description='QUOTE'/>
<value enum='T' description='SETTLEMENT_INSTRUCTIONS'/>
<value enum='V' description='MARKET_DATA_REQUEST'/>
<value enum='W' description='MARKET_DATA_SNAPSHOT_FULL_REFRESH'/>
<value enum='X' description='MARKET_DATA_INCREMENTAL_REFRESH'/>
<value enum='Y' description='MARKET_DATA_REQUEST_REJECT'/>
<value enum='Z' description='QUOTE_CANCEL'/>
<value enum='a' description='QUOTE_STATUS_REQUEST'/>
<value enum='b' description='MASS_QUOTE_ACKNOWLEDGEMENT'/>
<value enum='c' description='SECURITY_DEFINITION_REQUEST'/>
<value enum='d' description='SECURITY_DEFINITION'/>
<value enum='e' description='SECURITY_STATUS_REQUEST'/>
<value enum='f' description='SECURITY_STATUS'/>
<value enum='g' description='TRADING_SESSION_STATUS_REQUEST'/>
<value enum='h' description='TRADING_SESSION_STATUS'/>
<value enum='i' description='MASS_QUOTE'/>
<value enum='j' description='BUSINESS_MESSAGE_REJECT'/>
<value enum='k' description='BID_REQUEST'/>
<value enum='l' description='BID_RESPONSE'/>
<value enum='m' description='LIST_STRIKE_PRICE'/>
<value enum='n' description='XML_MESSAGE'/>
<value enum='o' description='REGISTRATION_INSTRUCTIONS'/>
<value enum='p' description='REGISTRATION_INSTRUCTIONS_RESPONSE'/>
<value enum='q' description='ORDER_MASS_CANCEL_REQUEST'/>
<value enum='r' description='ORDER_MASS_CANCEL_REPORT'/>
<value enum='s' description='NEW_ORDER_CROSS'/>
<value enum='t' description='CROSS_ORDER_CANCEL_REPLACE_REQUEST'/>
<value enum='u' description='CROSS_ORDER_CANCEL_REQUEST'/>
<value enum='v' description='SECURITY_TYPE_REQUEST'/>
<value enum='w' description='SECURITY_TYPES'/>
<value enum='x' description='SECURITY_LIST_REQUEST'/>
<value enum='y' description='SECURITY_LIST'/>
<value enum='z' description='DERIVATIVE_SECURITY_LIST_REQUEST'/>
<value enum='AA' description='DERIVATIVE_SECURITY_LIST'/>
<value enum='AB' description='NEW_ORDER_MULTILEG'/>
<value enum='AC' description='MULTILEG_ORDER_CANCEL_REPLACE'/>
<value enum='AD' description='TRADE_CAPTURE_REPORT_REQUEST'/>
<value enum='AE' description='TRADE_CAPTURE_REPORT'/>
<value enum='AF' description='ORDER_MASS_STATUS_REQUEST'/>
<value enum='AG' description='QUOTE_REQUEST_REJECT'/>
<value enum='AH' description='RFQ_REQUEST'/>
<value enum='AI' description='QUOTE_STATUS_REPORT'/>
<value enum='AJ' description='QUOTE_RESPONSE'/>
<value enum='AK' description='CONFIRMATION'/>
<value enum='AL' description='POSITION_MAINTENANCE_REQUEST'/>
<value enum='AM' description='POSITION_MAINTENANCE_REPORT'/>
<value enum='AN' description='REQUEST_FOR_POSITIONS'/>
<value enum='AO' description='REQUEST_FOR_POSITIONS_ACK'/>
<value enum='AP' description='POSITION_REPORT'/>
<value enum='AQ' description='TRADE_CAPTURE_REPORT_REQUEST_ACK'/>
<value enum='AR' description='TRADE_CAPTURE_REPORT_ACK'/>
<value enum='AS' description='ALLOCATION_REPORT'/>
<value enum='AT' description='ALLOCATION_REPORT_ACK'/>
<value enum='AU' description='CONFIRMATION_ACK'/>
<value enum='AV' description='SETTLEMENT_INSTRUCTION_REQUEST'/>
<value enum='AW' description='ASSIGNMENT_REPORT'/>
<value enum='AX' description='COLLATERAL_REQUEST'/>
<value enum='AY' description='COLLATERAL_ASSIGNMENT'/>
<value enum='AZ' description='COLLATERAL_RESPONSE'/>
<value enum='BA' description='COLLATERAL_REPORT'/>
<value enum='BB' description='COLLATERAL_INQUIRY'/>
<value enum='BC' description='NETWORK_STATUS_REQUEST'/>
<value enum='BD' description='NETWORK_STATUS_RESPONSE'/>
<value enum='BE' description='USER_REQUEST'/>
<value enum='BF' description='USER_RESPONSE'/>
<value enum='BG' description='COLLATERAL_INQUIRY_ACK'/>
<value enum='BH' description='CONFIRMATION_REQUEST'/>
<value enum='BI' description='TRADING_SESSION_LIST_REQUEST'/>
<value enum='BJ' description='TRADING_SESSION_LIST'/>
<value enum='BK' description='SECURITY_LIST_UPDATE_REPORT'/>
<value enum='BL' description='ADJUSTED_POSITION_REPORT'/>
<value enum='BM' description='ALLOCATION_INSTRUCTION_ALERT'/>
<value enum='BN' description='EXECUTION_ACKNOWLEDGEMENT'/>
<value enum='BO' description='CONTRARY_INTENTION_REPORT'/>
<value enum='BP' description='SECURITY_DEFINITION_UPDATE_REPORT'/>
</field>
<field name='NewSeqNo' number='36' type='SEQNUM'/>
<field name='PossDupFlag' number='43' type='BOOLEAN'/>
<field name='RefSeqNum' number='45' type='SEQNUM'/>
<field name='SenderCompID' number='49' type='STRING'/>
<field name='SenderSubID' number='50' type='STRING'/>
<field name='SendingTime' number='52' type='UTCTIMESTAMP'/>
<field name='TargetCompID' number='56' type='STRING'/>
<field name='TargetSubID' number='57' type='STRING'/>
<field name='Text' number='58' type='STRING'/>
<field name='Signature' number='89' type='DATA'/>
<field name='SecureDataLen' number='90' type='LENGTH'/>
<field name='SecureData' number='91' type='DATA'/>
<field name='SignatureLength' number='93' type='LENGTH'/>
<field name='RawDataLength' number='95' type='LENGTH'/>
<field name='RawData' number='96' type='DATA'/>
<field name='PossResend' number='97' type='BOOLEAN'/>
<field name='EncryptMethod' number='98' type='INT'>
<value description='NONE_OTHER' enum='0'/>
<value description='PKCS' enum='1'/>
<value description='DES' enum='2'/>
<value description='PKCS_DES' enum='3'/>
<value description='PGP_DES' enum='4'/>
<value description='PGP_DES_MD5' enum='5'/>
<value description='PEM_DES_MD5' enum='6'/>
</field>
<field name='HeartBtInt' number='108' type='INT'/>
<field name='TestReqID' number='112' type='STRING'/>
<field name='OnBehalfOfCompID' number='115' type='STRING'/>
<field name='OnBehalfOfSubID' number='116' type='STRING'/>
<field name='OrigSendingTime' number='122' type='UTCTIMESTAMP'/>
<field name='GapFillFlag' number='123' type='BOOLEAN'/>
<field name='DeliverToCompID' number='128' type='STRING'/>
<field name='DeliverToSubID' number='129' type='STRING'/>
<field name='ResetSeqNumFlag' number='141' type='BOOLEAN'/>
<field name='SenderLocationID' number='142' type='STRING'/>
<field name='TargetLocationID' number='143' type='STRING'/>
<field name='OnBehalfOfLocationID' number='144' type='STRING'/>
<field name='DeliverToLocationID' number='145' type='STRING'/>
<field name='XmlDataLen' number='212' type='LENGTH'/>
<field name='XmlData' number='213' type='DATA'/>
<field number='347' name='MessageEncoding' type='STRING'>
<value enum='ISO-2022-JP' description='ISO_2022_JP'/>
<value enum='EUC-JP' description='EUC_JP'/>
<value enum='SHIFT_JIS' description='SHIFT_JIS'/>
<value enum='UTF-8' description='UTF_8'/>
</field>
<field name='EncodedTextLen' number='354' type='LENGTH'/>
<field name='EncodedText' number='355' type='DATA'/>
<field name='LastMsgSeqNumProcessed' number='369' type='SEQNUM'/>
<field name='RefTagID' number='371' type='INT'/>
<field name='RefMsgType' number='372' type='STRING'/>
<field name='SessionRejectReason' number='373' type='INT'>
<value description='INVALID_TAG_NUMBER' enum='0'/>
<value description='REQUIRED_TAG_MISSING' enum='1'/>
<value description='SENDINGTIME_ACCURACY_PROBLEM' enum='10'/>
<value description='INVALID_MSGTYPE' enum='11'/>
<value description='XML_VALIDATION_ERROR' enum='12'/>
<value description='TAG_APPEARS_MORE_THAN_ONCE' enum='13'/>
<value description='TAG_SPECIFIED_OUT_OF_REQUIRED_ORDER' enum='14'/>
<value description='REPEATING_GROUP_FIELDS_OUT_OF_ORDER' enum='15'/>
<value description='INCORRECT_NUMINGROUP_COUNT_FOR_REPEATING_GROUP' enum='16'/>
<value description='NON_DATA_VALUE_INCLUDES_FIELD_DELIMITER' enum='17'/>
<value description='TAG_NOT_DEFINED_FOR_THIS_MESSAGE_TYPE' enum='2'/>
<value description='UNDEFINED_TAG' enum='3'/>
<value description='TAG_SPECIFIED_WITHOUT_A_VALUE' enum='4'/>
<value description='VALUE_IS_INCORRECT' enum='5'/>
<value description='INCORRECT_DATA_FORMAT_FOR_VALUE' enum='6'/>
<value description='DECRYPTION_PROBLEM' enum='7'/>
<value description='SIGNATURE_PROBLEM' enum='8'/>
<value description='COMPID_PROBLEM' enum='9'/>
<value description='OTHER' enum='99'/>
</field>
<field name='MaxMessageSize' number='383' type='LENGTH'/>
<field name='NoMsgTypes' number='384' type='NUMINGROUP'/>
<field name='MsgDirection' number='385' type='CHAR'>
<value description='RECEIVE' enum='R'/>
<value description='SEND' enum='S'/>
</field>
<field name='TestMessageIndicator' number='464' type='BOOLEAN'/>
<field name='Username' number='553' type='STRING'/>
<field name='Password' number='554' type='STRING'/>
<field name='NoHops' number='627' type='NUMINGROUP'/>
<field name='HopCompID' number='628' type='STRING'/>
<field name='HopSendingTime' number='629' type='UTCTIMESTAMP'/>
<field name='HopRefID' number='630' type='SEQNUM'/>
<field name='NextExpectedMsgSeqNum' number='789' type='SEQNUM'/>
<field name='ApplVerID' number='1128' type='STRING'>
<value description='FIX27' enum='0'/>
<value description='FIX30' enum='1'/>
<value description='FIX40' enum='2'/>
<value description='FIX41' enum='3'/>
<value description='FIX42' enum='4'/>
<value description='FIX43' enum='5'/>
<value description='FIX44' enum='6'/>
<value description='FIX50' enum='7'/>
<value description='FIX50SP1' enum='8'/>
<value description='FIX50SP2' enum='9'/>
</field>
<field name='CstmApplVerID' number='1129' type='STRING'/>
<field name='RefApplVerID' number='1130' type='STRING'/>
<field name='RefCstmApplVerID' number='1131' type='STRING'/>
<field name='DefaultApplVerID' number='1137' type='STRING'/>
</fields>
</fix>

View File

@ -36,11 +36,16 @@ type Service struct {
AuthorizedServices map[string]AuthorizedService `toml:"AuthorizedServices"` AuthorizedServices map[string]AuthorizedService `toml:"AuthorizedServices"`
APIBasePort string APIBasePort string
EnableJWTAuth bool // Enable JWT authentication for service-to-service communication EnableJWTAuth bool // Enable JWT authentication for service-to-service communication
// ServiceAPIKey is the shared secret that authenticates qbymarouter (and any
// other internal service) when posting to /qfixdpl/v1/quotes and
// /qfixdpl/v1/messages. Must match the DPLAPIKey configured on the caller.
ServiceAPIKey string
FIX FIXConfig FIX FIXConfig
} }
type FIXConfig struct { type FIXConfig struct {
SettingsFile string // path to fix.cfg file SettingsFile string // path to fix.cfg file
DataDictionaryFile string // path to FIX data dictionary XML (e.g. spec/FIX50SP2.xml)
} }
type ExtAuth struct { type ExtAuth struct {

View File

@ -328,6 +328,41 @@ func (cont *Controller) GetLogs(ctx *gin.Context) {
ctx.JSON(http.StatusOK, logs) ctx.JSON(http.StatusOK, logs)
} }
// AllMessages godoc
// @Summary List FIX application messages of the day after the caller's last-seen sequence
// @Description Returns today's FIX application messages (no admin: heartbeats/logon/logout/etc.) with MsgSeqNum greater than the caller's last-seen sequence per direction. "In" is the last MsgSeqNum the caller received on the IN side; "Out" is the same for OUT. Pass 0 to receive everything on that side. Sorted by CreatedAt ascending.
// @Tags fix
// @Accept json
// @Produce json
// @Param body body AllMessagesRequest true "API key and last-seen MsgSeqNum per direction"
// @Success 200 {array} domain.Message
// @Failure 400 {object} HTTPError
// @Failure 401 {object} HTTPError
// @Router /qfixdpl/v1/messages [post]
func (cont *Controller) AllMessages(ctx *gin.Context) {
setHeaders(ctx, cont.config)
var req AllMessagesRequest
if err := ctx.ShouldBindJSON(&req); err != nil {
ctx.JSON(http.StatusBadRequest, HTTPError{Error: err.Error()})
return
}
if !cont.checkServiceAPIKey(req.APIKey) {
ctx.JSON(http.StatusUnauthorized, HTTPError{Error: "Not allowed to perform this request"})
return
}
ctx.JSON(http.StatusOK, cont.tradeProvider.GetAllMessages(req.In, req.Out))
}
// checkServiceAPIKey returns true when the provided key matches the configured
// service-to-service shared secret. Empty configured key is always rejected
// to avoid open authentication when misconfigured.
func (cont *Controller) checkServiceAPIKey(key string) bool {
return cont.config.ServiceAPIKey != "" && key == cont.config.ServiceAPIKey
}
// GetPendingQuoteRequests godoc // GetPendingQuoteRequests godoc
// @Summary List pending QuoteRequests // @Summary List pending QuoteRequests
// @Description Returns all QuoteRequests received from TW that have not been quoted yet by the dealer // @Description Returns all QuoteRequests received from TW that have not been quoted yet by the dealer
@ -354,12 +389,19 @@ func (cont *Controller) GetPendingQuoteRequests(ctx *gin.Context) {
// @Failure 500 {object} HTTPError // @Failure 500 {object} HTTPError
// @Router /qfixdpl/v1/quotes [post] // @Router /qfixdpl/v1/quotes [post]
func (cont *Controller) SendQuote(ctx *gin.Context) { func (cont *Controller) SendQuote(ctx *gin.Context) {
setHeaders(ctx, cont.config)
var req SendQuoteRequest var req SendQuoteRequest
if err := ctx.ShouldBindJSON(&req); err != nil { if err := ctx.ShouldBindJSON(&req); err != nil {
ctx.JSON(http.StatusBadRequest, HTTPError{Error: err.Error()}) ctx.JSON(http.StatusBadRequest, HTTPError{Error: err.Error()})
return return
} }
if !cont.checkServiceAPIKey(req.APIKey) {
ctx.JSON(http.StatusUnauthorized, HTTPError{Error: "Not allowed to perform this request"})
return
}
price, err := decimal.NewFromString(req.Price) price, err := decimal.NewFromString(req.Price)
if err != nil { if err != nil {
ctx.JSON(http.StatusBadRequest, HTTPError{Error: "invalid price: " + err.Error()}) ctx.JSON(http.StatusBadRequest, HTTPError{Error: "invalid price: " + err.Error()})
@ -381,4 +423,3 @@ func (cont *Controller) SendQuote(ctx *gin.Context) {
ctx.JSON(http.StatusOK, Msg{Text: "Quote sent"}) ctx.JSON(http.StatusOK, Msg{Text: "Quote sent"})
} }

View File

@ -18,6 +18,13 @@ type Session struct {
} }
type SendQuoteRequest struct { type SendQuoteRequest struct {
APIKey string `json:"APIKey" binding:"required"`
QuoteReqID string `json:"QuoteReqID" binding:"required"` QuoteReqID string `json:"QuoteReqID" binding:"required"`
Price string `json:"Price" binding:"required" example:"99.6"` Price string `json:"Price" binding:"required" example:"99.6"`
} }
type AllMessagesRequest struct {
APIKey string
In int
Out int
}

View File

@ -24,7 +24,11 @@ func SetRoutes(api *API) {
qfixdpl.GET("/trades", cont.GetTrades) qfixdpl.GET("/trades", cont.GetTrades)
qfixdpl.GET("/trades/:quoteReqID/logs", cont.GetLogs) qfixdpl.GET("/trades/:quoteReqID/logs", cont.GetLogs)
qfixdpl.GET("/quote-requests", cont.GetPendingQuoteRequests) qfixdpl.GET("/quote-requests", cont.GetPendingQuoteRequests)
qfixdpl.POST("/quotes", cont.SendQuote)
// services group: API-key auth via body, no session cookie required.
services := v1.Group("/")
services.POST("/messages", cont.AllMessages)
services.POST("/quotes", cont.SendQuote)
backoffice := qfixdpl.Group("/backoffice") backoffice := qfixdpl.Group("/backoffice")
backoffice.Use(cont.BackOfficeUser) backoffice.Use(cont.BackOfficeUser)

View File

@ -22,6 +22,7 @@ type TradeProvider interface {
GetTrades() []domain.ListTrade GetTrades() []domain.ListTrade
GetPendingQuoteRequests() []domain.ListTrade GetPendingQuoteRequests() []domain.ListTrade
SendQuote(quoteReqID string, price decimal.Decimal) error SendQuote(quoteReqID string, price decimal.Decimal) error
GetAllMessages(inSeq, outSeq int) []domain.Message
} }
const RedisMaxIdle = 3000 // In ms const RedisMaxIdle = 3000 // In ms
@ -38,6 +39,10 @@ type Config struct {
AuthorizedServices map[string]app.AuthorizedService `toml:"AuthorizedServices"` AuthorizedServices map[string]app.AuthorizedService `toml:"AuthorizedServices"`
Port string Port string
EnableJWTAuth bool EnableJWTAuth bool
// ServiceAPIKey authenticates internal services (qbymarouter, etc.) calling
// /qfixdpl/v1/quotes and /qfixdpl/v1/messages. Compared against the APIKey
// field in the request body.
ServiceAPIKey string
} }
func New(userData app.UserDataProvider, storeInstance *store.Store, tradeProvider TradeProvider, config Config, notify domain.Notifier) *API { func New(userData app.UserDataProvider, storeInstance *store.Store, tradeProvider TradeProvider, config Config, notify domain.Notifier) *API {

172
src/client/fix/builder.go Normal file
View File

@ -0,0 +1,172 @@
package fix
import (
"fmt"
"quantex.com/qfixdpl/quickfix"
"quantex.com/qfixdpl/quickfix/datadictionary"
)
// BuildFieldMap walks a quickfix.FieldMap and produces an enriched FieldMap
// keyed by FixField (Name + Tag + Type) with values typed per the FIX
// data dictionary. Repeating groups become []FieldMap; nested groups recurse.
func BuildFieldMap(qfMap quickfix.FieldMap, dd *datadictionary.DataDictionary, fieldsByTag map[int]*datadictionary.FieldDef) FieldMap {
out := FieldMap{}
for _, t := range qfMap.Tags() {
tagInt := int(t)
rawValues := qfMap.RawValues(t)
if len(rawValues) == 0 {
continue
}
ff, fd := resolveField(dd, fieldsByTag, tagInt)
if ff.Type == "NUMINGROUP" {
if fd != nil && fd.IsGroup() {
out[ff] = buildGroup(rawValues[1:], dd, fd.Fields)
} else {
// Count tag with no group structure available (dictionary lookup
// failed, or quickfix didn't nest the inner fields). Emit an empty
// group rather than an int so the consumer's type expectation holds.
out[ff] = []FieldMap{}
}
continue
}
out[ff] = parseScalar(rawValues[0].Value(), ff.Type)
}
return out
}
// buildGroup splits a flat slice of TagValues (the inner values of a NUMINGROUP,
// excluding the count) into per-repetition FieldMaps. The first element of
// innerFields is the delimiter that marks the start of each repetition.
func buildGroup(tvs []quickfix.TagValue, dd *datadictionary.DataDictionary, innerFields []*datadictionary.FieldDef) []FieldMap {
if len(innerFields) == 0 || len(tvs) == 0 {
return nil
}
delimiter := innerFields[0].Tag()
fdByTag := make(map[int]*datadictionary.FieldDef, len(innerFields))
for _, f := range innerFields {
fdByTag[f.Tag()] = f
}
var (
repetitions []FieldMap
current FieldMap
)
i := 0
for i < len(tvs) {
tv := tvs[i]
tagInt := int(tv.Tag())
if tagInt == delimiter {
current = FieldMap{}
repetitions = append(repetitions, current)
}
if current == nil {
// Field appeared before the first delimiter; skip.
i++
continue
}
fd, known := fdByTag[tagInt]
if !known {
current[FixField{Name: fmt.Sprintf("tag_%d", tagInt), Tag: tagInt, Type: "STRING"}] = string(tv.Value())
i++
continue
}
if fd.IsGroup() {
nestedAllowed := allowedTags(fd.Fields)
j := i + 1
for j < len(tvs) && nestedAllowed[int(tvs[j].Tag())] {
j++
}
nested := buildGroup(tvs[i+1:j], dd, fd.Fields)
current[FixField{Name: fd.Name(), Tag: tagInt, Type: "NUMINGROUP"}] = nested
i = j
continue
}
current[FixField{Name: fd.Name(), Tag: tagInt, Type: fd.Type}] = parseScalar(tv.Value(), fd.Type)
i++
}
return repetitions
}
// resolveField looks up the FixField metadata (name + type) for a tag.
// Prefers the message-level FieldDef so group structure is preserved;
// falls back to the global FieldTypeByTag, then to a synthetic "tag_<N>" STRING.
func resolveField(dd *datadictionary.DataDictionary, fieldsByTag map[int]*datadictionary.FieldDef, tagInt int) (FixField, *datadictionary.FieldDef) {
if fieldsByTag != nil {
if fd, ok := fieldsByTag[tagInt]; ok {
return FixField{Name: fd.Name(), Tag: tagInt, Type: fd.Type}, fd
}
}
if dd != nil {
if ft, ok := dd.FieldTypeByTag[tagInt]; ok {
return FixField{Name: ft.Name(), Tag: tagInt, Type: ft.Type}, nil
}
}
return FixField{Name: fmt.Sprintf("tag_%d", tagInt), Tag: tagInt, Type: "STRING"}, nil
}
// parseScalar converts raw FIX bytes into the Go type expected by the consumer
// (GetKeyValue): bool for BOOLEAN, int for INT/SEQNUM, time.Time for UTCTIMESTAMP,
// string for everything else.
func parseScalar(raw []byte, fixType string) interface{} {
s := string(raw)
switch fixType {
case "INT", "SEQNUM", "LENGTH":
var v quickfix.FIXInt
if err := v.Read(raw); err == nil {
return int(v)
}
return s
case "BOOLEAN":
var v quickfix.FIXBoolean
if err := v.Read(raw); err == nil {
return bool(v)
}
return s
case "UTCTIMESTAMP":
var v quickfix.FIXUTCTimestamp
if err := v.Read(raw); err == nil {
return v.Time
}
return s
default:
return s
}
}
// allowedTags returns the set of all tags valid inside a group (including
// nested-group counts and their descendants), used to detect where a flat
// nested-group slice ends within its parent.
func allowedTags(fields []*datadictionary.FieldDef) map[int]bool {
out := make(map[int]bool, len(fields))
var visit func(fs []*datadictionary.FieldDef)
visit = func(fs []*datadictionary.FieldDef) {
for _, f := range fs {
out[f.Tag()] = true
if f.IsGroup() {
visit(f.Fields)
}
}
}
visit(fields)
return out
}

106
src/client/fix/dict.go Normal file
View File

@ -0,0 +1,106 @@
package fix
import (
"log/slog"
"time"
"quantex.com/qfixdpl/src/common/tracerr"
)
// FixField identifies a single FIX field by its dictionary metadata.
type FixField struct {
Name string
Tag int
Type string
}
// FieldValue is the typed value for a FIX field.
type FieldValue interface{}
// FieldMap is the enriched representation of a FIX FieldMap (header, body, or trailer).
type FieldMap map[FixField]FieldValue
// GetMap converts a FieldMap into a JSON-friendly map keyed by field name.
func GetMap(fieldMap FieldMap) map[string]interface{} {
result := map[string]interface{}{}
for f, v := range fieldMap {
slog.Info("try to parse fieldMap: %+v, value: %+v", f, v)
k, val := GetKeyValue(f, v)
result[k] = val
}
return result
}
//nolint:funlen,gocyclo,cyclop //it's long but easy to read
func GetKeyValue(f FixField, value FieldValue) (key string, val interface{}) {
key = f.Name
switch f.Type {
case "NUMINGROUP":
var groups []map[string]interface{}
fMapLst, ok := value.([]FieldMap)
if !ok {
err := tracerr.Errorf("could not parse as []FieldMap, value: %+v, key: %s", value, key)
slog.Error(err.Error())
}
for _, fieldMap := range fMapLst {
groups = append(groups, GetMap(fieldMap))
}
val = groups
case "BOOLEAN":
b, ok := value.(bool)
if !ok {
err := tracerr.Errorf("could not parse as bool, value: %+v, key: %s", value, key)
slog.Error(err.Error())
}
val = b
case "INT", "LENGTH", "SEQNUM":
i, ok := value.(int)
if !ok {
err := tracerr.Errorf("could not parse as int, value: %+v, key: %s", value, key)
slog.Error(err.Error())
}
val = i
case "UTCTIMESTAMP":
t, ok := value.(time.Time)
if !ok {
err := tracerr.Errorf("could not parse as Time, value: %+v, key: %s", value, key)
slog.Error(err.Error())
}
val = t
case "STRING":
s, ok := value.(string)
if !ok {
err := tracerr.Errorf("could not parse as string, value: %+v, key: %s", value, key)
slog.Error(err.Error())
}
val = s
case "CHAR":
s, ok := value.(string)
if !ok {
err := tracerr.Errorf("could not parse as string, value: %+v, key: %s", value, key)
slog.Error(err.Error())
}
val = s
default:
s, ok := value.(string)
if !ok {
err := tracerr.Errorf("could not parse as string, value: %+v, key: %s", value, key)
slog.Error(err.Error())
}
val = s
}
return key, val
}

View File

@ -3,13 +3,17 @@ package fix
import ( import (
"log/slog" "log/slog"
"os" "os"
"sort"
"strings" "strings"
"sync" "sync"
"time" "time"
"github.com/rs/zerolog/log"
uuid "github.com/satori/go.uuid"
"github.com/shopspring/decimal" "github.com/shopspring/decimal"
"quantex.com/qfixdpl/quickfix" "quantex.com/qfixdpl/quickfix"
"quantex.com/qfixdpl/quickfix/datadictionary"
"quantex.com/qfixdpl/quickfix/gen/enum" "quantex.com/qfixdpl/quickfix/gen/enum"
"quantex.com/qfixdpl/quickfix/gen/field" "quantex.com/qfixdpl/quickfix/gen/field"
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/executionack" "quantex.com/qfixdpl/quickfix/gen/fix50sp2/executionack"
@ -19,6 +23,7 @@ import (
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoterequest" "quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoterequest"
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoteresponse" "quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoteresponse"
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/quotestatusreport" "quantex.com/qfixdpl/quickfix/gen/fix50sp2/quotestatusreport"
"quantex.com/qfixdpl/quickfix/gen/tag"
filelog "quantex.com/qfixdpl/quickfix/log/file" filelog "quantex.com/qfixdpl/quickfix/log/file"
"quantex.com/qfixdpl/quickfix/store/file" "quantex.com/qfixdpl/quickfix/store/file"
"quantex.com/qfixdpl/src/app" "quantex.com/qfixdpl/src/app"
@ -27,18 +32,10 @@ import (
) )
type listTrade struct { type listTrade struct {
QuoteReqID string QuoteRequest domain.FixMessageJSON
ListID string
Symbol string
SecurityIDSrc enum.SecurityIDSource
Currency string
Side enum.Side
OrderQty decimal.Decimal
SettlDate string
Price decimal.Decimal
OwnerTraderID string
SessionID quickfix.SessionID SessionID quickfix.SessionID
Quoted bool Quoted bool
Price decimal.Decimal
} }
// Manager wraps the QuickFIX initiator and implements domain.FIXSender. // Manager wraps the QuickFIX initiator and implements domain.FIXSender.
@ -49,15 +46,19 @@ type Manager struct {
sessions map[string]quickfix.SessionID sessions map[string]quickfix.SessionID
tradesMu sync.RWMutex tradesMu sync.RWMutex
trades map[string]*listTrade trades map[string]*listTrade
messagesMu sync.RWMutex
messages []domain.Message
store domain.PersistenceStore store domain.PersistenceStore
notify domain.Notifier notify domain.Notifier
cfg app.FIXConfig cfg app.FIXConfig
dict *datadictionary.DataDictionary
} }
func NewManager(cfg app.FIXConfig, store domain.PersistenceStore, notify domain.Notifier) *Manager { func NewManager(cfg app.FIXConfig, store domain.PersistenceStore, notify domain.Notifier) *Manager {
return &Manager{ return &Manager{
sessions: make(map[string]quickfix.SessionID), sessions: make(map[string]quickfix.SessionID),
trades: make(map[string]*listTrade), trades: make(map[string]*listTrade),
messages: make([]domain.Message, 0),
store: store, store: store,
notify: notify, notify: notify,
cfg: cfg, cfg: cfg,
@ -76,11 +77,25 @@ func (m *Manager) Start() error {
fixApp.onRawMessage = m.handleRawMessage fixApp.onRawMessage = m.handleRawMessage
m.app = fixApp m.app = fixApp
dict, err := datadictionary.Parse(m.cfg.DataDictionaryFile)
if err != nil {
err = tracerr.Errorf("error parsing FIX data dictionary %q: %w", m.cfg.DataDictionaryFile, err)
slog.Error(err.Error())
return err
}
m.dict = dict
if err := m.loadActiveTrades(); err != nil { if err := m.loadActiveTrades(); err != nil {
err = tracerr.Errorf("failed to load active trades from DB, starting with empty state: %w", err) err = tracerr.Errorf("failed to load active trades from DB, starting with empty state: %w", err)
slog.Error(err.Error()) slog.Error(err.Error())
} }
if err := m.loadTodayMessages(); err != nil {
err = tracerr.Errorf("failed to load today messages from DB, starting with empty list: %w", err)
slog.Error(err.Error())
}
f, err := os.Open(m.cfg.SettingsFile) f, err := os.Open(m.cfg.SettingsFile)
if err != nil { if err != nil {
err = tracerr.Errorf("error opening FIX settings file %q: %w", m.cfg.SettingsFile, err) err = tracerr.Errorf("error opening FIX settings file %q: %w", m.cfg.SettingsFile, err)
@ -191,52 +206,50 @@ func (m *Manager) sendExecutionAck(orderID, clOrdID, execID string, sessionID qu
return quickfix.SendToTarget(bn, sessionID) return quickfix.SendToTarget(bn, sessionID)
} }
// handleQuoteRequest auto-responds to an incoming QuoteRequest with a QuoteStatusReport (acknowledge) followed by a Quote at price 99.6. // handleQuoteRequest auto-responds to an incoming QuoteRequest with a QuoteStatusReport (acknowledge).
// The Quote (35=S) is sent later via the REST endpoint when the dealer prices it.
func (m *Manager) handleQuoteRequest(msg quoterequest.QuoteRequest, sessionID quickfix.SessionID) { func (m *Manager) handleQuoteRequest(msg quoterequest.QuoteRequest, sessionID quickfix.SessionID) {
quoteReqID, err := msg.GetQuoteReqID() parsed := parseQuoteRequest(msg, m.dict)
if err != nil { quoteReqID := parsed.QuoteReqID
err := tracerr.Errorf("handleQuoteRequest: missing QuoteReqID: %w", err)
slog.Error(err.Error())
if quoteReqID == "" {
m.notify.SendMsg(
domain.MessageChannelError,
"quoteReqID missing in quote request",
domain.MessageStatusWarning,
nil,
)
err := tracerr.Errorf("handleQuoteRequest, missing QuoteReqID, quoteRequest: %+v", parsed)
slog.Error(err.Error())
return return
} }
// Validate LST_ prefix for List Trading flow. // Validate LST_ prefix for List Trading flow.
if !strings.HasPrefix(quoteReqID, "LST_") { if !strings.HasPrefix(quoteReqID, "LST_") {
m.notify.SendMsg(
domain.MessageChannelError,
"quoteReqID ("+quoteReqID+") missing LST_ prefix",
domain.MessageStatusWarning,
nil,
)
slog.Warn("handleQuoteRequest: QuoteReqID missing LST_ prefix, ignoring", "quoteReqID", quoteReqID) slog.Warn("handleQuoteRequest: QuoteReqID missing LST_ prefix, ignoring", "quoteReqID", quoteReqID)
return return
} }
var ( bodyKeys := make([]string, 0, len(parsed.Body))
symbol, currency, ownerTraderID, settlDate, listID, negotiationType string for k := range parsed.Body {
side enum.Side bodyKeys = append(bodyKeys, k)
secIDSource enum.SecurityIDSource
orderQty decimal.Decimal
)
relatedSyms, relErr := msg.GetNoRelatedSym()
if relErr == nil && relatedSyms.Len() > 0 {
sym := relatedSyms.Get(0)
symbol, _ = sym.GetSecurityID()
secIDSource, _ = sym.GetSecurityIDSource()
currency, _ = sym.GetCurrency()
side, _ = sym.GetSide()
ownerTraderID, _ = sym.GetOwnerTraderID()
orderQty, _ = sym.GetOrderQty()
settlDate, _ = sym.GetSettlDate()
listID, _ = sym.GetListID()
negotiationType, _ = sym.GetNegotiationType()
} }
slog.Info("handleQuoteRequest: parsed body keys", "quoteReqID", quoteReqID, "keys", bodyKeys)
if listID == "" { relSym := firstGroup(parsed.Body, "NoRelatedSym")
slog.Warn("handleQuoteRequest: missing ListID", "quoteReqID", quoteReqID) relSymKeys := make([]string, 0, len(relSym))
return for k := range relSym {
relSymKeys = append(relSymKeys, k)
} }
slog.Info("handleQuoteRequest: NoRelatedSym keys", "quoteReqID", quoteReqID, "keys", relSymKeys)
if negotiationType != "RFQ" { ownerTraderID := getString(relSym, "OwnerTraderID")
slog.Warn("handleQuoteRequest: unexpected NegotiationType", "quoteReqID", quoteReqID, "negotiationType", negotiationType)
return
}
// Send QuoteStatusReport (35=AI) to acknowledge the inquiry. // Send QuoteStatusReport (35=AI) to acknowledge the inquiry.
if ackErr := m.sendQuoteStatusReport(quoteReqID, ownerTraderID, sessionID); ackErr != nil { if ackErr := m.sendQuoteStatusReport(quoteReqID, ownerTraderID, sessionID); ackErr != nil {
@ -246,37 +259,14 @@ func (m *Manager) handleQuoteRequest(msg quoterequest.QuoteRequest, sessionID qu
} }
slog.Info("QuoteStatusReport sent", "quoteReqID", quoteReqID) slog.Info("QuoteStatusReport sent", "quoteReqID", quoteReqID)
sIDSource := enum.SecurityIDSource_ISIN_NUMBER
if secIDSource == enum.SecurityIDSource_CUSIP {
sIDSource = enum.SecurityIDSource_CUSIP
}
// Store trade state as pending; Quote (35=S) is sent later via REST endpoint. // Store trade state as pending; Quote (35=S) is sent later via REST endpoint.
m.tradesMu.Lock() m.tradesMu.Lock()
m.trades[quoteReqID] = &listTrade{ m.trades[quoteReqID] = &listTrade{
QuoteReqID: quoteReqID, QuoteRequest: parsed,
ListID: listID,
Symbol: symbol,
SecurityIDSrc: sIDSource,
Currency: currency,
Side: side,
OrderQty: orderQty,
SettlDate: settlDate,
OwnerTraderID: ownerTraderID,
SessionID: sessionID, SessionID: sessionID,
Quoted: false, Quoted: false,
} }
m.tradesMu.Unlock() m.tradesMu.Unlock()
// Persist structured message (outside mutex).
m.persistMessage(quoteReqID, parseQuoteRequest(msg))
// Persist outgoing QuoteStatusReport.
m.persistMessage(quoteReqID, buildOutgoingMessageJSON("AI", quoteReqID, map[string]interface{}{
"QuoteReqID": quoteReqID,
"QuoteStatus": string(enum.QuoteStatus_ACCEPTED),
"OwnerTraderID": ownerTraderID,
}))
} }
// handleQuoteAck handles an incoming QuoteAck (35=CW). // handleQuoteAck handles an incoming QuoteAck (35=CW).
@ -285,16 +275,10 @@ func (m *Manager) handleQuoteAck(msg quoteack.QuoteAck, sessionID quickfix.Sessi
status, _ := msg.GetQuoteAckStatus() status, _ := msg.GetQuoteAckStatus()
text, _ := msg.GetText() text, _ := msg.GetText()
m.persistMessage(quoteReqID, parseQuoteAck(msg))
if status != enum.QuoteAckStatus_ACCEPTED { if status != enum.QuoteAckStatus_ACCEPTED {
err := tracerr.Errorf("handleQuoteAck: quote rejected by TW (quoteReqID=%s, quoteAckStatus=%s, text=%s)", quoteReqID, string(status), text) err := tracerr.Errorf("handleQuoteAck: quote rejected by TW (quoteReqID=%s, quoteAckStatus=%s, text=%s)", quoteReqID, string(status), text)
slog.Error(err.Error()) slog.Error(err.Error())
m.tradesMu.Lock()
delete(m.trades, quoteReqID)
m.tradesMu.Unlock()
return return
} }
@ -330,22 +314,9 @@ func (m *Manager) handleQuoteResponse(msg quoteresponse.QuoteResponse, sessionID
} }
slog.Info("QuoteResponse ACK sent", "quoteReqID", quoteReqID, "quoteRespID", quoteRespID) slog.Info("QuoteResponse ACK sent", "quoteReqID", quoteReqID, "quoteRespID", quoteRespID)
// Persist incoming QuoteResponse.
m.persistMessage(quoteReqID, parseQuoteResponse(msg))
// Persist outgoing ACK.
m.persistMessage(quoteReqID, buildOutgoingMessageJSON("AI", quoteReqID, map[string]interface{}{
"QuoteReqID": quoteReqID,
"QuoteRespID": quoteRespID,
"QuoteStatus": string(enum.QuoteStatus_ACCEPTED),
}))
// _TRDSUMM is the final message — clean up the trade. // _TRDSUMM is the final message — clean up the trade.
if isTrdSumm { if isTrdSumm {
slog.Info("Trade summary received, cleaning up", "quoteReqID", quoteReqID) slog.Info("Trade summary received, cleaning up", "quoteReqID", quoteReqID)
m.tradesMu.Lock()
delete(m.trades, quoteReqID)
m.tradesMu.Unlock()
} }
} }
@ -384,25 +355,10 @@ func (m *Manager) handleExecutionReport(msg executionreport.ExecutionReport, ses
slog.Info("Trade summary received from TW, cleaning up", slog.Info("Trade summary received from TW, cleaning up",
"execID", execID, "clOrdID", clOrdID, "ordStatus", string(ordStatus), "listID", listID) "execID", execID, "clOrdID", clOrdID, "ordStatus", string(ordStatus), "listID", listID)
m.tradesMu.Lock()
delete(m.trades, clOrdID)
m.tradesMu.Unlock()
case execType == enum.ExecType_TRADE: case execType == enum.ExecType_TRADE:
slog.Info("Trade result received from TW", slog.Info("Trade result received from TW",
"execID", execID, "clOrdID", clOrdID, "ordStatus", string(ordStatus), "listID", listID) "execID", execID, "clOrdID", clOrdID, "ordStatus", string(ordStatus), "listID", listID)
} }
// Persist incoming ExecutionReport.
m.persistMessage(clOrdID, parseExecutionReport(msg))
// Persist outgoing ExecutionAck.
m.persistMessage(clOrdID, buildOutgoingMessageJSON("BN", clOrdID, map[string]interface{}{
"OrderID": orderID,
"ExecID": execID,
"ClOrdID": clOrdID,
"ExecAckStatus": string(enum.ExecAckStatus_ACCEPTED),
}))
} }
// handleExecutionAck handles an incoming ExecutionAck (35=BN) from TW. // handleExecutionAck handles an incoming ExecutionAck (35=BN) from TW.
@ -430,27 +386,18 @@ func (m *Manager) GetPendingQuoteRequests() []domain.ListTrade {
pending := make([]domain.ListTrade, 0) pending := make([]domain.ListTrade, 0)
for _, t := range m.trades { for _, t := range m.trades {
if !t.Quoted {
pending = append(pending, toDomainListTrade(t)) pending = append(pending, toDomainListTrade(t))
} }
}
return pending return pending
} }
func toDomainListTrade(t *listTrade) domain.ListTrade { func toDomainListTrade(t *listTrade) domain.ListTrade {
return domain.ListTrade{ out := domain.ListTrade{
QuoteReqID: t.QuoteReqID, QuoteRequest: t.QuoteRequest,
ListID: t.ListID,
Symbol: t.Symbol,
SecurityIDSrc: string(t.SecurityIDSrc),
Currency: t.Currency,
Side: string(t.Side),
OrderQty: t.OrderQty.String(),
SettlDate: t.SettlDate,
Price: t.Price.String(),
OwnerTraderID: t.OwnerTraderID,
} }
return out
} }
// SendQuote builds and sends a Quote (35=S) for an existing pending QuoteRequest at the given price. // SendQuote builds and sends a Quote (35=S) for an existing pending QuoteRequest at the given price.
@ -463,12 +410,6 @@ func (m *Manager) SendQuote(quoteReqID string, price decimal.Decimal) error {
slog.Error(err.Error()) slog.Error(err.Error())
return err return err
} }
if t.Quoted {
m.tradesMu.Unlock()
err := tracerr.Errorf("SendQuote: quote already sent for quoteReqID %s", quoteReqID)
slog.Error(err.Error())
return err
}
sessionID := t.SessionID sessionID := t.SessionID
if sessionID == (quickfix.SessionID{}) { if sessionID == (quickfix.SessionID{}) {
@ -481,15 +422,19 @@ func (m *Manager) SendQuote(quoteReqID string, price decimal.Decimal) error {
} }
} }
symbol := t.Symbol symbol := getString(t.QuoteRequest.Body, "SecurityID")
sIDSource := t.SecurityIDSrc sIDSource := enum.SecurityIDSource(getString(t.QuoteRequest.Body, "SecurityIDSource"))
currency := t.Currency currency := getString(t.QuoteRequest.Body, "Currency")
side := t.Side side := enum.Side(getString(t.QuoteRequest.Body, "Side"))
orderQty := t.OrderQty orderQty := getDecimal(t.QuoteRequest.Body, "OrderQty")
settlDate := t.SettlDate settlDate := getString(t.QuoteRequest.Body, "SettlDate")
ownerTraderID := t.OwnerTraderID ownerTraderID := getString(t.QuoteRequest.Body, "OwnerTraderID")
m.tradesMu.Unlock() m.tradesMu.Unlock()
if sIDSource != enum.SecurityIDSource_CUSIP {
sIDSource = enum.SecurityIDSource_ISIN_NUMBER
}
quoteID := quoteReqID quoteID := quoteReqID
q := quote.New( q := quote.New(
field.NewQuoteID(quoteID), field.NewQuoteID(quoteID),
@ -545,20 +490,48 @@ func (m *Manager) SendQuote(quoteReqID string, price decimal.Decimal) error {
slog.Info("Quote sent", "quoteReqID", quoteReqID, "quoteID", quoteID, "symbol", symbol, "price", price.String()) slog.Info("Quote sent", "quoteReqID", quoteReqID, "quoteID", quoteID, "symbol", symbol, "price", price.String())
m.persistMessage(quoteReqID, buildOutgoingMessageJSON("S", quoteReqID, map[string]interface{}{
"QuoteReqID": quoteReqID,
"QuoteID": quoteID,
"Symbol": symbol,
"Side": string(side),
"Price": price.String(),
"OrderQty": orderQty.String(),
"Currency": currency,
"SettlDate": settlDate,
}))
return nil return nil
} }
// firstGroup returns the first repetition of a NUMINGROUP field, or nil.
func firstGroup(body map[string]any, name string) map[string]any {
if body == nil {
return nil
}
groups, ok := body[name].([]map[string]any)
if !ok || len(groups) == 0 {
return nil
}
return groups[0]
}
// getString reads a string value from a body map, tolerating nil maps and missing keys.
func getString(body map[string]any, name string) string {
if body == nil {
return ""
}
if v, ok := body[name].(string); ok {
return v
}
return ""
}
// getDecimal reads a decimal value from a body map. Numeric FIX types come through
// as strings (e.g. "10000"); INT-typed counts may be int. Both are accepted.
func getDecimal(body map[string]any, name string) decimal.Decimal {
if body == nil {
return decimal.Decimal{}
}
switch v := body[name].(type) {
case string:
d, _ := decimal.NewFromString(v)
return d
case int:
return decimal.NewFromInt(int64(v))
}
return decimal.Decimal{}
}
func (m *Manager) anyActiveSessionID() quickfix.SessionID { func (m *Manager) anyActiveSessionID() quickfix.SessionID {
m.sessionsMu.RLock() m.sessionsMu.RLock()
defer m.sessionsMu.RUnlock() defer m.sessionsMu.RUnlock()
@ -568,7 +541,11 @@ func (m *Manager) anyActiveSessionID() quickfix.SessionID {
return quickfix.SessionID{} return quickfix.SessionID{}
} }
// handleRawMessage persists raw FIX message strings to the logs table. // handleRawMessage is the single ingest path for every application-level FIX message
// (admin messages — Logon, Logout, Heartbeat, TestRequest, ResendRequest, SequenceReset —
// go through ToAdmin/FromAdmin and never reach this callback). It persists the raw
// envelope to the logs table, builds a structured Message and saves it to
// the messages table, and appends to the in-memory list.
func (m *Manager) handleRawMessage(direction string, msg *quickfix.Message) { func (m *Manager) handleRawMessage(direction string, msg *quickfix.Message) {
quoteReqID := extractIdentifier(msg) quoteReqID := extractIdentifier(msg)
@ -579,17 +556,77 @@ func (m *Manager) handleRawMessage(direction string, msg *quickfix.Message) {
err = tracerr.Errorf("failed to persist raw log: %w", err) err = tracerr.Errorf("failed to persist raw log: %w", err)
slog.Error(err.Error()) slog.Error(err.Error())
} }
}
// persistMessage saves a structured FIX message to the messages table. msgTypeBytes, _ := msg.Header.GetBytes(tag.MsgType)
func (m *Manager) persistMessage(quoteReqID string, fixJSON domain.FixMessageJSON) { msgType := string(msgTypeBytes)
if err := m.store.SaveMessage(domain.TradeMessage{ senderCompID, msgSeqNum, sendingTime := extractHeaderMeta(msg)
QuoteReqID: quoteReqID,
fixJSON := buildFixMessageJSON(direction, msgType, quoteReqID, msg, m.dict)
stored := domain.Message{
ID: uuid.NewV4().String(),
SenderCompID: senderCompID,
MsgSeqNum: msgSeqNum,
SendingTime: sendingTime,
CreatedAt: time.Now(),
JMessage: fixJSON, JMessage: fixJSON,
}); err != nil { }
err = tracerr.Errorf("failed to persist message (msgType=%s, quoteReqID=%s): %w", fixJSON.MsgType, quoteReqID, err)
if err := m.store.SaveMessage(stored); err != nil {
err = tracerr.Errorf("failed to persist message (msgType=%s, quoteReqID=%s): %w", msgType, quoteReqID, err)
slog.Error(err.Error()) slog.Error(err.Error())
} }
m.messagesMu.Lock()
m.messages = append(m.messages, stored)
m.messagesMu.Unlock()
}
// GetAllMessages returns today's FIX application messages with MsgSeqNum greater than
// the caller's last-seen sequence per direction (inSeq for IN, outSeq for OUT), sorted
// ascending by CreatedAt. Passing 0 for either cursor returns all messages on that side.
func (m *Manager) GetAllMessages(inSeq, outSeq int) []domain.Message {
m.messagesMu.RLock()
log.Info().Msgf("request received, inSeq: %d, outSeq: %d", inSeq, outSeq)
filtered := make([]domain.Message, 0, len(m.messages))
for _, msg := range m.messages {
switch msg.JMessage.Direction {
case "IN":
if msg.MsgSeqNum > inSeq {
filtered = append(filtered, msg)
}
case "OUT":
if msg.MsgSeqNum > outSeq {
filtered = append(filtered, msg)
}
}
}
m.messagesMu.RUnlock()
sort.Slice(filtered, func(i, j int) bool { return filtered[i].CreatedAt.Before(filtered[j].CreatedAt) })
log.Info().Msgf("messages sent: %d", len(filtered))
return filtered
}
// loadTodayMessages rebuilds the in-memory message list from today's rows in the DB.
// Must be called before the FIX initiator starts so live ingest doesn't race with replay.
func (m *Manager) loadTodayMessages() error {
messages, err := m.store.GetTodayMessages()
if err != nil {
return err
}
m.messagesMu.Lock()
m.messages = messages
m.messagesMu.Unlock()
slog.Info("today messages loaded", "count", len(messages))
return nil
} }
// loadActiveTrades reconstructs active trades from today's messages in the database. // loadActiveTrades reconstructs active trades from today's messages in the database.
@ -602,90 +639,30 @@ func (m *Manager) loadActiveTrades() error {
activeTrades := make(map[string]*listTrade) activeTrades := make(map[string]*listTrade)
for _, msg := range messages { for _, msg := range messages {
quoteReqID := msg.JMessage.QuoteReqID
switch msg.JMessage.MsgType { switch msg.JMessage.MsgType {
case "R": // QuoteRequest -> trade is born case "R": // QuoteRequest -> trade is born
if !strings.HasPrefix(msg.QuoteReqID, "LST_") { if !strings.HasPrefix(quoteReqID, "LST_") {
continue continue
} }
body := msg.JMessage.Body relSym := firstGroup(msg.JMessage.Body, "NoRelatedSym")
if getString(relSym, "NegotiationType") != "RFQ" {
nt, _ := body["NegotiationType"].(string) continue
if nt != "RFQ" { }
if getString(relSym, "ListID") == "" {
continue continue
} }
listID, _ := body["ListID"].(string) activeTrades[quoteReqID] = &listTrade{
if listID == "" { QuoteRequest: msg.JMessage,
continue
} }
trade := &listTrade{
QuoteReqID: msg.QuoteReqID,
ListID: listID,
}
if v, ok := body["SecurityID"].(string); ok {
trade.Symbol = v
}
if v, ok := body["SecurityIDSource"].(string); ok {
trade.SecurityIDSrc = enum.SecurityIDSource(v)
}
if v, ok := body["Currency"].(string); ok {
trade.Currency = v
}
if v, ok := body["Side"].(string); ok {
trade.Side = enum.Side(v)
}
if v, ok := body["OrderQty"].(string); ok {
trade.OrderQty, _ = decimal.NewFromString(v)
}
if v, ok := body["SettlDate"].(string); ok {
trade.SettlDate = v
}
if v, ok := body["OwnerTraderID"].(string); ok {
trade.OwnerTraderID = v
}
activeTrades[msg.QuoteReqID] = trade
case "S": // Outgoing Quote — dealer has already quoted this trade case "S": // Outgoing Quote — dealer has already quoted this trade
if t, ok := activeTrades[msg.QuoteReqID]; ok { if t, ok := activeTrades[quoteReqID]; ok {
t.Quoted = true t.Quoted = true
if v, ok := msg.JMessage.Body["Price"].(string); ok { t.Price = getDecimal(msg.JMessage.Body, "Price")
t.Price, _ = decimal.NewFromString(v)
}
}
case "CW": // QuoteAck — if rejected, trade is dead
body := msg.JMessage.Body
quoteAckStatus, _ := body["QuoteAckStatus"].(string)
if quoteAckStatus != string(enum.QuoteAckStatus_ACCEPTED) {
delete(activeTrades, msg.QuoteReqID)
}
case "AJ": // QuoteResponse — _TRDSUMM means trade is done (flow 8.6)
body := msg.JMessage.Body
quoteRespID, _ := body["QuoteRespID"].(string)
if strings.HasSuffix(quoteRespID, "_TRDSUMM") {
delete(activeTrades, msg.QuoteReqID)
}
case "8": // ExecutionReport — _TRDSUMM means trade is done (flow 8.4)
body := msg.JMessage.Body
execID, _ := body["ExecID"].(string)
clOrdID, _ := body["ClOrdID"].(string)
if strings.Contains(execID, "_TRDSUMM") {
delete(activeTrades, clOrdID)
} }
} }
} }

View File

@ -4,196 +4,47 @@ import (
"time" "time"
"quantex.com/qfixdpl/quickfix" "quantex.com/qfixdpl/quickfix"
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/executionreport" "quantex.com/qfixdpl/quickfix/datadictionary"
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoteack"
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoterequest" "quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoterequest"
"quantex.com/qfixdpl/quickfix/gen/fix50sp2/quoteresponse"
"quantex.com/qfixdpl/quickfix/gen/tag" "quantex.com/qfixdpl/quickfix/gen/tag"
"quantex.com/qfixdpl/src/domain" "quantex.com/qfixdpl/src/domain"
) )
func extractHeader(msg *quickfix.Message) map[string]interface{} { // buildFixMessageJSON walks the full FIX message (header + body + trailer)
header := make(map[string]interface{}) // using the data dictionary and returns a fully populated FixMessageJSON.
func buildFixMessageJSON(direction, msgType, quoteReqID string, msg *quickfix.Message, dd *datadictionary.DataDictionary) domain.FixMessageJSON {
var (
headerFields map[int]*datadictionary.FieldDef
trailerFields map[int]*datadictionary.FieldDef
bodyFields map[int]*datadictionary.FieldDef
)
if v, err := msg.Header.GetBytes(tag.BeginString); err == nil { if dd != nil {
header["BeginString"] = string(v) if dd.Header != nil {
headerFields = dd.Header.Fields
} }
if v, err := msg.Header.GetBytes(tag.MsgType); err == nil { if dd.Trailer != nil {
header["MsgType"] = string(v) trailerFields = dd.Trailer.Fields
} }
if v, err := msg.Header.GetBytes(tag.SenderCompID); err == nil { if md, ok := dd.Messages[msgType]; ok {
header["SenderCompID"] = string(v) bodyFields = md.Fields
}
if v, err := msg.Header.GetBytes(tag.TargetCompID); err == nil {
header["TargetCompID"] = string(v)
}
if v, err := msg.Header.GetBytes(tag.MsgSeqNum); err == nil {
header["MsgSeqNum"] = string(v)
}
if v, err := msg.Header.GetBytes(tag.SendingTime); err == nil {
header["SendingTime"] = string(v)
}
return header
}
func parseQuoteRequest(msg quoterequest.QuoteRequest) domain.FixMessageJSON {
quoteReqID, _ := msg.GetQuoteReqID()
body := map[string]interface{}{"QuoteReqID": quoteReqID}
if relSyms, err := msg.GetNoRelatedSym(); err == nil && relSyms.Len() > 0 {
sym := relSyms.Get(0)
if v, e := sym.GetSecurityID(); e == nil {
body["SecurityID"] = v
}
if v, e := sym.GetSecurityIDSource(); e == nil {
body["SecurityIDSource"] = string(v)
}
if v, e := sym.GetCurrency(); e == nil {
body["Currency"] = v
}
if v, e := sym.GetSide(); e == nil {
body["Side"] = string(v)
}
if v, e := sym.GetOrderQty(); e == nil {
body["OrderQty"] = v.String()
}
if v, e := sym.GetSettlDate(); e == nil {
body["SettlDate"] = v
}
if v, e := sym.GetListID(); e == nil {
body["ListID"] = v
}
if v, e := sym.GetOwnerTraderID(); e == nil {
body["OwnerTraderID"] = v
}
if v, e := sym.GetNegotiationType(); e == nil {
body["NegotiationType"] = v
} }
} }
return domain.FixMessageJSON{ return domain.FixMessageJSON{
Direction: "IN", Direction: direction,
MsgType: "R", MsgType: msgType,
QuoteReqID: quoteReqID, QuoteReqID: quoteReqID,
Header: extractHeader(msg.Message), Header: GetMap(BuildFieldMap(msg.Header.FieldMap, dd, headerFields)),
Body: body, Body: GetMap(BuildFieldMap(msg.Body.FieldMap, dd, bodyFields)),
Trailer: GetMap(BuildFieldMap(msg.Trailer.FieldMap, dd, trailerFields)),
ReceiveTime: time.Now(), ReceiveTime: time.Now(),
} }
} }
func parseQuoteAck(msg quoteack.QuoteAck) domain.FixMessageJSON { func parseQuoteRequest(msg quoterequest.QuoteRequest, dd *datadictionary.DataDictionary) domain.FixMessageJSON {
quoteReqID, _ := msg.GetQuoteReqID() quoteReqID, _ := msg.GetQuoteReqID()
body := map[string]interface{}{"QuoteReqID": quoteReqID} return buildFixMessageJSON("IN", "R", quoteReqID, msg.Message, dd)
if v, e := msg.GetQuoteID(); e == nil {
body["QuoteID"] = v
}
if v, e := msg.GetQuoteAckStatus(); e == nil {
body["QuoteAckStatus"] = string(v)
}
if v, e := msg.GetText(); e == nil {
body["Text"] = v
}
return domain.FixMessageJSON{
Direction: "IN",
MsgType: "CW",
QuoteReqID: quoteReqID,
Header: extractHeader(msg.Message),
Body: body,
ReceiveTime: time.Now(),
}
}
func parseQuoteResponse(msg quoteresponse.QuoteResponse) domain.FixMessageJSON {
quoteReqID, _ := msg.GetQuoteReqID()
body := map[string]interface{}{"QuoteReqID": quoteReqID}
if v, e := msg.GetQuoteRespID(); e == nil {
body["QuoteRespID"] = v
}
if v, e := msg.GetQuoteRespType(); e == nil {
body["QuoteRespType"] = string(v)
}
if v, e := msg.GetSide(); e == nil {
body["Side"] = string(v)
}
if v, e := msg.GetPrice(); e == nil {
body["Price"] = v.String()
}
if v, e := msg.GetOrderQty(); e == nil {
body["OrderQty"] = v.String()
}
if v, e := msg.GetClOrdID(); e == nil {
body["ClOrdID"] = v
}
return domain.FixMessageJSON{
Direction: "IN",
MsgType: "AJ",
QuoteReqID: quoteReqID,
Header: extractHeader(msg.Message),
Body: body,
ReceiveTime: time.Now(),
}
}
func parseExecutionReport(msg executionreport.ExecutionReport) domain.FixMessageJSON {
clOrdID, _ := msg.GetClOrdID()
body := map[string]interface{}{"ClOrdID": clOrdID}
if v, e := msg.GetExecID(); e == nil {
body["ExecID"] = v
}
if v, e := msg.GetOrderID(); e == nil {
body["OrderID"] = v
}
if v, e := msg.GetExecType(); e == nil {
body["ExecType"] = string(v)
}
if v, e := msg.GetOrdStatus(); e == nil {
body["OrdStatus"] = string(v)
}
if v, e := msg.GetListID(); e == nil {
body["ListID"] = v
}
if v, e := msg.GetSide(); e == nil {
body["Side"] = string(v)
}
if v, e := msg.GetSymbol(); e == nil {
body["Symbol"] = v
}
if v, e := msg.GetSecurityID(); e == nil {
body["SecurityID"] = v
}
if v, e := msg.GetCurrency(); e == nil {
body["Currency"] = v
}
if v, e := msg.GetPrice(); e == nil {
body["Price"] = v.String()
}
if v, e := msg.GetLastPx(); e == nil {
body["LastPx"] = v.String()
}
if v, e := msg.GetLastQty(); e == nil {
body["LastQty"] = v.String()
}
if v, e := msg.GetOrderQty(); e == nil {
body["OrderQty"] = v.String()
}
if v, e := msg.GetSettlDate(); e == nil {
body["SettlDate"] = v
}
return domain.FixMessageJSON{
Direction: "IN",
MsgType: "8",
QuoteReqID: clOrdID,
Header: extractHeader(msg.Message),
Body: body,
ReceiveTime: time.Now(),
}
} }
// extractIdentifier extracts the trade identifier from a parsed FIX message. // extractIdentifier extracts the trade identifier from a parsed FIX message.
@ -218,12 +69,17 @@ func extractIdentifier(msg *quickfix.Message) string {
return "" return ""
} }
func buildOutgoingMessageJSON(msgType, quoteReqID string, body map[string]interface{}) domain.FixMessageJSON { // extractHeaderMeta reads SenderCompID (49), MsgSeqNum (34) and SendingTime (52)
return domain.FixMessageJSON{ // from a quickfix.Message header. Returns zero values when a field is absent.
Direction: "OUT", func extractHeaderMeta(msg *quickfix.Message) (senderCompID string, msgSeqNum int, sendingTime time.Time) {
MsgType: msgType, if s, err := msg.Header.GetString(tag.SenderCompID); err == nil {
QuoteReqID: quoteReqID, senderCompID = s
Body: body,
ReceiveTime: time.Now(),
} }
if n, err := msg.Header.GetInt(tag.MsgSeqNum); err == nil {
msgSeqNum = n
}
if t, err := msg.Header.GetTime(tag.SendingTime); err == nil {
sendingTime = t
}
return
} }

View File

@ -1,11 +1,12 @@
CREATE TABLE IF NOT EXISTS qfixdpl_messages ( CREATE TABLE IF NOT EXISTS qfixdpl_messages (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), id UUID PRIMARY KEY,
quote_req_id TEXT NOT NULL, sender_comp_id TEXT NOT NULL,
msg_seq_num BIGINT NOT NULL,
j_message JSONB NOT NULL, j_message JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() created_at TIMESTAMPTZ NOT NULL
); );
CREATE INDEX IF NOT EXISTS idx_messages_quote_req_id ON qfixdpl_messages(quote_req_id);
CREATE INDEX IF NOT EXISTS idx_messages_created_at ON qfixdpl_messages(created_at); CREATE INDEX IF NOT EXISTS idx_messages_created_at ON qfixdpl_messages(created_at);
CREATE INDEX IF NOT EXISTS idx_messages_session ON qfixdpl_messages(sender_comp_id, msg_seq_num);
CREATE TABLE IF NOT EXISTS qfixdpl_logs ( CREATE TABLE IF NOT EXISTS qfixdpl_logs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), id UUID PRIMARY KEY DEFAULT gen_random_uuid(),

View File

@ -2,6 +2,7 @@ package store
import ( import (
"encoding/json" "encoding/json"
"strconv"
"strings" "strings"
"time" "time"
@ -9,15 +10,20 @@ import (
"quantex.com/qfixdpl/src/domain" "quantex.com/qfixdpl/src/domain"
) )
func (p *Store) SaveMessage(msg domain.TradeMessage) error { func (p *Store) SaveMessage(msg domain.Message) error {
jsonBytes, err := json.Marshal(msg.JMessage) jsonBytes, err := json.Marshal(msg.JMessage)
if err != nil { if err != nil {
return tracerr.Errorf("error marshaling j_message: %w", err) return tracerr.Errorf("error marshaling j_message: %w", err)
} }
_, err = p.db.Exec( _, err = p.db.Exec(
"INSERT INTO qfixdpl_messages (quote_req_id, j_message) VALUES ($1, $2)", `INSERT INTO qfixdpl_messages (id, sender_comp_id, msg_seq_num, j_message, created_at)
msg.QuoteReqID, string(jsonBytes), VALUES ($1, $2, $3, $4, $5)`,
msg.ID,
msg.SenderCompID,
strconv.Itoa(msg.MsgSeqNum),
string(jsonBytes),
msg.CreatedAt.UTC().Format(time.RFC3339Nano),
) )
if err != nil { if err != nil {
return tracerr.Errorf("error inserting message: %w", err) return tracerr.Errorf("error inserting message: %w", err)
@ -41,25 +47,29 @@ func (p *Store) SaveLog(entry domain.LogEntry) error {
return nil return nil
} }
func (p *Store) GetTodayMessages() ([]domain.TradeMessage, error) { func (p *Store) GetTodayMessages() ([]domain.Message, error) {
rows, err := p.db.Query( rows, err := p.db.Query(
"SELECT id, quote_req_id, j_message, created_at FROM qfixdpl_messages WHERE created_at >= current_date ORDER BY created_at ASC", `SELECT id, sender_comp_id, msg_seq_num, j_message, created_at
FROM qfixdpl_messages
WHERE created_at >= current_date
ORDER BY created_at ASC`,
) )
if err != nil { if err != nil {
return nil, tracerr.Errorf("error querying today messages: %w", err) return nil, tracerr.Errorf("error querying today messages: %w", err)
} }
defer rows.Close() defer rows.Close()
var messages []domain.TradeMessage var messages []domain.Message
for rows.Next() { for rows.Next() {
var ( var (
id, quoteReqID string id, senderCompID string
msgSeqNum int
jMessageRaw []byte jMessageRaw []byte
createdAt time.Time createdAt time.Time
) )
if err := rows.Scan(&id, &quoteReqID, &jMessageRaw, &createdAt); err != nil { if err := rows.Scan(&id, &senderCompID, &msgSeqNum, &jMessageRaw, &createdAt); err != nil {
return nil, tracerr.Errorf("error scanning message row: %w", err) return nil, tracerr.Errorf("error scanning message row: %w", err)
} }
@ -68,11 +78,22 @@ func (p *Store) GetTodayMessages() ([]domain.TradeMessage, error) {
return nil, tracerr.Errorf("error unmarshaling j_message: %w", err) return nil, tracerr.Errorf("error unmarshaling j_message: %w", err)
} }
messages = append(messages, domain.TradeMessage{ sendingTime, _ := jMessage.Header["SendingTime"].(time.Time)
if sendingTime.IsZero() {
if s, ok := jMessage.Header["SendingTime"].(string); ok {
if t, parseErr := time.Parse(time.RFC3339Nano, s); parseErr == nil {
sendingTime = t
}
}
}
messages = append(messages, domain.Message{
ID: id, ID: id,
QuoteReqID: quoteReqID, SenderCompID: senderCompID,
JMessage: jMessage, MsgSeqNum: msgSeqNum,
SendingTime: sendingTime,
CreatedAt: createdAt, CreatedAt: createdAt,
JMessage: jMessage,
}) })
} }

View File

@ -51,6 +51,7 @@ func Runner(cfg app.Config) error {
External: cfg.External, External: cfg.External,
AuthorizedServices: cfg.AuthorizedServices, AuthorizedServices: cfg.AuthorizedServices,
EnableJWTAuth: cfg.EnableJWTAuth, EnableJWTAuth: cfg.EnableJWTAuth,
ServiceAPIKey: cfg.ServiceAPIKey,
} }
api := rest.New(userData, appStore, fixManager, apiConfig, notify) api := rest.New(userData, appStore, fixManager, apiConfig, notify)

View File

@ -5,16 +5,7 @@ import "time"
// ListTrade es la representacion exportada de un trade de List Trading. // ListTrade es la representacion exportada de un trade de List Trading.
type ListTrade struct { type ListTrade struct {
QuoteReqID string QuoteRequest FixMessageJSON `json:"quote_request"`
ListID string
Symbol string
SecurityIDSrc string
Currency string
Side string
OrderQty string
SettlDate string
Price string
OwnerTraderID string
} }
// FixMessageJSON es la representacion estructurada de un mensaje FIX para almacenamiento. // FixMessageJSON es la representacion estructurada de un mensaje FIX para almacenamiento.
@ -22,17 +13,21 @@ type FixMessageJSON struct {
Direction string `json:"direction"` Direction string `json:"direction"`
MsgType string `json:"msg_type"` MsgType string `json:"msg_type"`
QuoteReqID string `json:"quote_req_id"` QuoteReqID string `json:"quote_req_id"`
Header map[string]interface{} `json:"header"` Header map[string]any `json:"header"`
Body map[string]interface{} `json:"body"` Body map[string]any `json:"body"`
Trailer map[string]any `json:"trailer"`
ReceiveTime time.Time `json:"receive_time"` ReceiveTime time.Time `json:"receive_time"`
} }
// TradeMessage es una fila de qfixdpl_messages. // Message es una fila de qfixdpl_messages, con la metadata del header FIX hoisted
type TradeMessage struct { // para que los consumidores puedan ordenar/filtrar sin parsear el JSON.
type Message struct {
ID string `json:"id"` ID string `json:"id"`
QuoteReqID string `json:"quote_req_id"` SenderCompID string `json:"sender_comp_id"`
JMessage FixMessageJSON `json:"j_message"` MsgSeqNum int `json:"msg_seq_num"`
SendingTime time.Time `json:"sending_time"`
CreatedAt time.Time `json:"created_at"` CreatedAt time.Time `json:"created_at"`
JMessage FixMessageJSON `json:"j_message"`
} }
// LogEntry es el DTO para insertar/actualizar un log crudo en qfixdpl_logs. // LogEntry es el DTO para insertar/actualizar un log crudo en qfixdpl_logs.
@ -48,8 +43,8 @@ type Logs struct {
// PersistenceStore define la interfaz de persistencia. // PersistenceStore define la interfaz de persistencia.
type PersistenceStore interface { type PersistenceStore interface {
SaveMessage(msg TradeMessage) error SaveMessage(msg Message) error
SaveLog(entry LogEntry) error SaveLog(entry LogEntry) error
GetTodayMessages() ([]TradeMessage, error) GetTodayMessages() ([]Message, error)
GetLogsByQuoteReqID(quoteReqID string) (Logs, error) GetLogsByQuoteReqID(quoteReqID string) (Logs, error)
} }